Structured Streaming + Kafka

Ý tưởng cốt lõi của Structured Streaming: coi luồng dữ liệu như một bảng vô hạn (unbounded table) mà cứ có dữ liệu mới là một dòng được append vào cuối bảng. Bạn viết query y hệt như khi xử lý batch — select, filter, groupBy — còn Spark sẽ tự chạy lặp lại query đó theo từng micro-batch mỗi khi có dữ liệu mới đến.

Nói cách khác, bạn không phải học một API mới. Sự khác biệt nằm ở chỗ nguồn vào/ra là stream chứ không phải file tĩnh, và job chạy liên tục thay vì kết thúc sau một lần.

Để mở một stream từ Kafka, dùng spark.readStream với format kafka và khai báo bootstrap server cùng topic cần đọc:

df = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "dbserver1.inventory.customers")
    .option("startingOffsets", "earliest")
    .load())

Các thành phần:

• readStream — mở một DataFrame dạng stream thay vì đọc tĩnh.
• format("kafka") — chỉ định nguồn là Kafka source.
• subscribe — tên topic muốn đọc (Debezium đặt tên dạng server.database.table).
• startingOffsets — đọc từ đâu: earliest (từ message cũ nhất còn lưu) hoặc latest (chỉ message mới từ lúc chạy).

Kafka source luôn trả về một schema cố định — bất kể nội dung message là gì:

Cột value là binary, nên trước hết phải CAST sang string rồi mới parse JSON. Khai báo một schema mô tả cấu trúc JSON mong muốn, rồi dùng from_json để bóc tách:

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, LongType

schema = StructType([
    StructField("id", LongType()),
    StructField("name", StringType()),
    StructField("email", StringType()),
])

parsed = (df.selectExpr("CAST(value AS STRING) AS json")
    .select(F.from_json("json", schema).alias("data"))
    .select("data.*"))

selectExpr("CAST(value AS STRING) ...") đổi binary thành chuỗi JSON; from_json(cột, schema) chuyển chuỗi đó thành một struct theo schema; cuối cùng select("data.*") bung struct thành các cột phẳng.

Ví dụ schema lồng đúng cấu trúc Debezium, rồi chọn payload.after.* và payload.op:

after_schema = StructType([
    StructField("id", LongType()),
    StructField("name", StringType()),
    StructField("email", StringType()),
])
payload_schema = StructType([
    StructField("op", StringType()),
    StructField("after", after_schema),
])
envelope = StructType([StructField("payload", payload_schema)])

cdc = (df.selectExpr("CAST(value AS STRING) AS json")
    .select(F.from_json("json", envelope).alias("m"))
    .select("m.payload.op", "m.payload.after.*")
    .filter("op IN ('c','u')"))   # bỏ qua delete

Để khởi chạy stream, gọi writeStream rồi .start(). Khi học, ghi ra console là tiện nhất — thấy ngay từng micro-batch in ra màn hình:

query = (parsed.writeStream
    .format("console")
    .outputMode("append")
    .option("truncate", False)
    .start())

query.awaitTermination()

outputMode quyết định Spark ghi cái gì sau mỗi micro-batch:

Các sink hay dùng:

Với stream, ta thường muốn gom dữ liệu theo cửa sổ thời gian — ví dụ đếm số event mỗi 5 phút. Dùng F.window trong groupBy, kết hợp withWatermark để xử lý dữ liệu đến trễ:

from pyspark.sql import functions as F

agg = (parsed
    .withWatermark("ts", "10 minutes")
    .groupBy(F.window("ts", "5 minutes"), "type")
    .count())

window("ts", "5 minutes") tạo các cửa sổ tumbling (không chồng nhau) rộng 5 phút dựa trên cột thời gian ts; mỗi event rơi vào đúng một cửa sổ và được đếm theo cửa sổ + cột type.

withWatermark("ts", "10 minutes") báo cho Spark biết: dữ liệu có thể đến trễ tối đa 10 phút. Nhờ đó Spark vừa vẫn cộng đúng những event đến muộn trong giới hạn này, vừa được phép dọn state cũ của các cửa sổ đã quá hạn (nếu không, state sẽ phình mãi).

query = (parsed.writeStream
    .format("console")
    .outputMode("append")
    .option("checkpointLocation", "/home/jovyan/work/data/chk/job1")
    .start())

• Bật cdc-lab, rồi bật spark-lab kèm compose Kafka:
  docker compose -f docker-compose.yml -f docker-compose.kafka.yml up -d
• Tạo hoặc sửa một row trong database nguồn (ví dụ bảng customers) để Debezium bắt thay đổi và phát message CDC lên topic Kafka.
• Chạy notebook readStream → console như ở phần trên; mỗi micro-batch sẽ in message vừa phát ra màn hình.
• Thử thêm window aggregation (đếm theo cửa sổ thời gian) và quan sát kết quả cập nhật dần.

• Đọc một topic CDC bất kỳ từ cdc-lab, parse JSON Debezium và in ra op cùng các cột trong after.* bằng sink console.
• Đếm số sự kiện theo loại thao tác (c / u / d) theo cửa sổ 1 phút, dùng F.window + withWatermark.
• Ghi stream đã parse xuống Parquet với outputMode("append") và một checkpointLocation riêng; kiểm tra file Parquet được tạo ra trong folder data.