Mục tiêu

Tổng hợp toàn bộ kiến thức từ các bài học trước để triển khai một dự án thực tế.
Áp dụng Flink trong xử lý dữ liệu streaming từ Kafka, lưu trữ và phân tích dữ liệu với Elasticsearch và Kibana.
Thực hiện tối ưu hóa hiệu năng, giám sát, và xử lý lỗi trong môi trường production.

Nội dung chính

1. Phân tích yêu cầu dự án

Case Study: Clickstream Analysis:
- Hệ thống cần thu thập dữ liệu hành vi người dùng từ website (click, search, add-to-cart).
- Phân tích dữ liệu để:
  - Đếm số lần click theo user và thời gian.
  - Phát hiện hành vi bất thường (click liên tục trong thời gian ngắn).
  - Tạo dashboard hiển thị các phân tích trong Kibana.
Luồng dữ liệu:
- Source: Kafka.
- Processing: Flink.
- Sink: Elasticsearch.

2. Thiết kế kiến trúc

Pipeline tổng quan:
1. Kafka nhận dữ liệu từ ứng dụng website.
2. Flink xử lý và phát hiện mẫu dữ liệu (CEP).
3. Ghi dữ liệu đã xử lý vào Elasticsearch.
4. Hiển thị dashboard trong Kibana.

3. Xây dựng pipeline Flink

Cài đặt Kafka Source và Elasticsearch Sink

Kafka Source:

  Properties kafkaProps = new Properties();
  kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
  kafkaProps.setProperty("group.id", "clickstream-group");

  DataStream<String> kafkaSource = env.addSource(
      new FlinkKafkaConsumer<>("clickstream", new SimpleStringSchema(), kafkaProps)
  );

Elasticsearch Sink:

  ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
      List.of(new HttpHost("localhost", 9200, "http")),
      (element, ctx, indexer) -> {
          Map<String, String> json = new HashMap<>();
          json.put("user", element);
          indexer.add(Requests.indexRequest().index("clickstream").source(json));
      }
  );
  kafkaSource.addSink(esSinkBuilder.build());

Phân tích và phát hiện hành vi bất thường

Định nghĩa pattern FlinkCEP:

Phát hiện click liên tục trong thời gian ngắn:

  Pattern<ClickEvent, ?> clickPattern = Pattern
      .<ClickEvent>begin("start")
      .where(click -> click.getType().equals("CLICK"))
      .times(5).within(Time.seconds(10));

Ánh xạ dữ liệu từ Kafka vào pattern:

  PatternStream<ClickEvent> patternStream = CEP.pattern(kafkaSource, clickPattern);

  DataStream<String> alerts = patternStream.select(
      (PatternSelectFunction<ClickEvent, String>) pattern -> {
          return "Suspicious activity detected: " + pattern.get("start");
      }
  );

4. Tối ưu hóa hiệu năng

Cấu hình parallelism:
```
  env.setParallelism(4);
```
Checkpointing:
- Đảm bảo exactly-once:
```
  env.enableCheckpointing(5000);
```

State Backend:

Sử dụng RocksDB để lưu state:

  env.setStateBackend(new EmbeddedRocksDBStateBackend());

5. Hiển thị dashboard trong Kibana

Tạo index pattern trong Kibana:
- Truy cập Kibana UI: http://localhost:5601.
- Tạo index pattern clickstream để hiển thị dữ liệu.
Tạo visualization:
- Visualization 1: Số lần click theo user.
- Visualization 2: Biểu đồ thời gian hiển thị hành vi bất thường.

Ví dụ minh hoạ: Mã nguồn hoàn chỉnh

public class ClickstreamAnalysis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Kafka Source
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "clickstream-group");

        DataStream<String> kafkaSource = env.addSource(
            new FlinkKafkaConsumer<>("clickstream", new SimpleStringSchema(), kafkaProps)
        );

        // CEP Pattern
        Pattern<ClickEvent, ?> clickPattern = Pattern
            .<ClickEvent>begin("start")
            .where(click -> click.getType().equals("CLICK"))
            .times(5).within(Time.seconds(10));

        PatternStream<ClickEvent> patternStream = CEP.pattern(kafkaSource, clickPattern);

        DataStream<String> alerts = patternStream.select(
            (PatternSelectFunction<ClickEvent, String>) pattern -> {
                return "Suspicious activity detected: " + pattern.get("start");
            }
        );

        // Elasticsearch Sink
        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
            List.of(new HttpHost("localhost", 9200, "http")),
            (element, ctx, indexer) -> {
                Map<String, String> json = new HashMap<>();
                json.put("alert", element);
                indexer.add(Requests.indexRequest().index("clickstream-alerts").source(json));
            }
        );
        alerts.addSink(esSinkBuilder.build());

        env.execute("Clickstream Analysis");
    }
}

Kết luận & Hướng mở rộng

Kết luận

Dự án này giúp bạn hiểu rõ cách kết hợp Kafka, Flink, Elasticsearch, và Kibana để xây dựng hệ thống phân tích dữ liệu streaming thực tế.
Các bước từ thiết kế, triển khai, đến tối ưu hóa đều đã được áp dụng.

Hướng mở rộng

Tích hợp Prometheus và Grafana để giám sát hiệu năng pipeline.
Thêm phân tích hành vi nâng cao (ví dụ: dự đoán bằng Machine Learning với FlinkML).
Sử dụng Flink SQL để thực hiện các phép toán phức tạp trên dữ liệu streaming.

Bài 15: Thực chiến Dự án End-to-End