Spark 6 — Structured Streaming

Từ batch sang streaming, nhưng vẫn là một DataFrame

Ở các bài trước của series, chúng ta xử lý dữ liệu theo lô (batch): đọc một tập file hoặc một bảng cố định, biến đổi bằng Spark SQL, rồi ghi kết quả ra. Tập dữ liệu có điểm đầu và điểm cuối rõ ràng — job chạy xong là kết thúc.

Nhưng nhiều bài toán ngân hàng không có "điểm cuối". Giao dịch thẻ, log đăng nhập, sự kiện từ core banking chảy về liên tục 24/7. Nếu cứ mỗi giờ chạy một job batch để tổng hợp, bạn luôn trễ ít nhất một giờ so với thực tế — quá muộn để cảnh báo gian lận hay giám sát hạn mức realtime.

Structured Streaming là engine xử lý luồng của Spark, xây trên chính Spark SQL engine. Ý tưởng cốt lõi rất đẹp: coi một luồng dữ liệu vô tận như một bảng đang được append liên tục (unbounded table). Mỗi khi có dữ liệu mới tới, các dòng mới được thêm vào cuối "bảng đầu vào" đó. Truy vấn bạn viết về mặt logic chạy trên toàn bộ bảng, nhưng Spark thực thi tăng dần (incremental) để chỉ tính phần mới.

Hệ quả thực tiễn: cùng một DataFrame API như batch — cùng select, filter, groupBy, join. Khác biệt gần như duy nhất là dùng spark.readStream thay cho spark.read, và df.writeStream thay cho df.write. Việc chuyển logic batch sang streaming vì thế rất nhẹ nhàng — lợi thế lớn so với viết lại bằng một API luồng hoàn toàn khác.

Lưu ý sandbox: môi trường minh hoạ của tài liệu này dùng PostgreSQL, không có Spark cluster. Toàn bộ mã PySpark bên dưới là minh hoạ để đọc hiểu, không chạy trực tiếp trong sandbox.

Mô hình unbounded table

Hãy hình dung ba khối:

• Input table — bảng vô tận, mỗi trigger được append thêm các dòng mới đến từ nguồn.
• Result table — kết quả của truy vấn bạn định nghĩa, cập nhật sau mỗi lần Spark xử lý một lô dữ liệu mới.
• Output (sink) — phần của Result table được ghi ra đích, theo output mode bạn chọn.

Mỗi chu kỳ xử lý, Spark làm ba việc: đọc offset mới từ nguồn, cập nhật Result table tăng dần, rồi ghi phần chênh lệch ra sink. Toàn bộ tiến độ (đã đọc tới đâu, state hiện tại) được lưu bền vững để khôi phục sau sự cố — nói kỹ ở mục checkpoint.

Micro-batch vs continuous

Mặc định, Structured Streaming chạy theo micro-batch: gom dữ liệu đến trong một khoảng ngắn thành một lô nhỏ, xử lý như một job Spark tí hon, rồi lặp lại. Độ trễ end-to-end thường cỡ trăm mili-giây đến vài giây — đủ tốt cho đa số bài toán, đồng thời tận dụng trọn engine batch đã tối ưu (Catalyst, Tungsten) và fault tolerance mạnh.

Chế độ Continuous Processing (vẫn thử nghiệm và còn nhiều giới hạn ở Spark 3.5) nhắm tới độ trễ mili-giây bằng cách chạy task dài hạn thay vì chia lô. Thực tế production phần lớn vẫn dùng micro-batch; chỉ cân nhắc continuous khi thật sự cần độ trễ cực thấp. Bài này tập trung vào micro-batch.

Nguồn và đích

Nguồn (source) phổ biến: Kafka (quan trọng nhất trong thực tế), file (đọc file mới xuất hiện trong một thư mục — CSV/JSON/Parquet), rate (sinh dữ liệu giả để test), và socket (chỉ để thử nghiệm, không fault-tolerant).

Đích (sink): Kafka, file, console (in ra để debug), memory (bảng tạm để query), và foreachBatch — sink linh hoạt nhất, cho phép bạn nhận DataFrame của từng micro-batch và tự quyết định ghi đi đâu (JDBC vào PostgreSQL, MERGE vào Delta, gọi API...).

Kafka là cặp đôi tự nhiên của Structured Streaming, nối tiếp series Kafka — Kiến trúc. Đọc bằng format("kafka"), chỉ ra kafka.bootstrap.servers và subscribe. Cần nhớ: giá trị message trả về là binary (key, value kiểu bytes), nên hầu như luôn phải CAST sang string rồi parse JSON/Avro để lấy schema thật.

Event time vs processing time

Đây là phân biệt nền tảng của mọi hệ xử lý luồng nghiêm túc.

• Processing time — thời điểm Spark xử lý một sự kiện. Phụ thuộc tốc độ máy, độ trễ mạng, độ tải; không ổn định.
• Event time — thời điểm sự kiện thực sự xảy ra, được nhúng trong chính dữ liệu (ví dụ trường txn_time do máy POS ghi lúc quẹt thẻ).

Bài toán tổng hợp có ý nghĩa nghiệp vụ gần như luôn cần event time. "Có bao nhiêu giao dịch trong khung 10:00–10:05" phải dựa vào lúc giao dịch xảy ra, không phải lúc Spark tình cờ đọc được nó.

Và đây là chỗ khó: dữ liệu đến trễ (late data). Một giao dịch xảy ra lúc 10:04 nhưng do mất sóng, đến 10:12 mới đẩy lên Kafka. Nếu ta đã "chốt sổ" cửa sổ 10:00–10:05 lúc 10:06, sự kiện trễ này rơi vào đâu? Bỏ qua thì sai số; chờ mãi thì không bao giờ chốt được và state phình vô hạn. Watermark sinh ra để giải quyết mâu thuẫn này.

Watermark — chốt kết quả và giải phóng state

Watermark là một ngưỡng do bạn khai báo: "Tôi chấp nhận chờ dữ liệu trễ tối đa X thời gian; quá đó thì bỏ." Khai báo bằng withWatermark("event_time", "10 minutes").

Cơ chế: Spark theo dõi event time lớn nhất đã thấy. Watermark = (max event time đã thấy) − (ngưỡng trễ). Một cửa sổ được finalize (chốt và có thể phát output ở append mode) khi watermark vượt qua điểm cuối cửa sổ. Sự kiện đến trễ hơn watermark hiện tại bị loại bỏ khỏi phép tổng hợp có state.

Watermark phục vụ hai mục đích cùng lúc:

1. Tính đúng đắn có kiểm soát — quyết định khi nào một kết quả tổng hợp coi như hoàn chỉnh để phát ra.
2. Giới hạn state — khi cửa sổ đã qua watermark, state tương ứng được giải phóng khỏi state store. Không có watermark, aggregation theo cửa sổ phải giữ state của mọi cửa sổ mãi mãi, dẫn tới tràn bộ nhớ.

Đánh đổi rõ ràng: watermark dài bắt được nhiều dữ liệu trễ hơn nhưng chốt chậm và state to; ngắn thì phát nhanh, state gọn, nhưng bỏ nhiều sự kiện trễ. Chọn theo phân phối độ trễ thực tế của nguồn.

Windowing trên event time

Muốn tổng hợp một dòng vô tận, bạn phải cắt nó thành các cửa sổ (window) theo event time:

• Tumbling window — cửa sổ cố định, không chồng lấn, liền kề nhau. Ví dụ mỗi 5 phút: [10:00,10:05), [10:05,10:10).... Mỗi sự kiện thuộc đúng một cửa sổ. Đây là loại dùng nhiều nhất.
• Sliding window — cửa sổ có độ dài cố định nhưng trượt theo bước nhỏ hơn, nên chồng lấn. Ví dụ cửa sổ 10 phút trượt mỗi 5 phút → một sự kiện có thể thuộc hai cửa sổ. Dùng cho đường trung bình trượt.
• Session window — cửa sổ động theo hoạt động: gom các sự kiện gần nhau (cách nhau dưới một khoảng "gap") thành một phiên; im lặng quá gap thì phiên đóng lại. Hợp cho phân tích phiên người dùng.

Cú pháp dùng hàm window(col, "5 minutes") (tumbling) hoặc window(col, "10 minutes", "5 minutes") (sliding) bên trong groupBy. Thực tế luôn kết hợp withWatermark trước khi groupBy(window(...)).

Output modes — khi nào dùng gì

Output mode quyết định phần nào của Result table được ghi ra sink mỗi trigger:

• append (mặc định) — chỉ ghi các dòng mới được thêm và không còn thay đổi nữa. Với aggregation, một cửa sổ chỉ thành "bất biến" khi đã qua watermark, nên append mode với windowed aggregation bắt buộc phải có watermark và kết quả chỉ phát ra sau khi cửa sổ chốt. Phù hợp ghi vào sink chỉ-append (file, Kafka).
• update — mỗi trigger ghi các dòng có thay đổi so với lần trước (kể cả cửa sổ đang cập nhật dở). Cho kết quả sớm và cập nhật dần; hợp với sink hỗ trợ upsert (ví dụ ghi qua foreachBatch rồi MERGE).
• complete — ghi lại toàn bộ Result table mỗi trigger. Chỉ khả thi khi kết quả đủ nhỏ (state không được giải phóng vì luôn phải xuất đầy đủ). Dùng cho bảng tổng hợp nhỏ cần hiển thị đầy đủ mỗi lần.

Không phải mode nào cũng hợp mọi truy vấn — ví dụ complete mode chỉ dùng được với aggregation. Chọn mode theo bản chất sink và độ trễ chấp nhận được.

Stateful processing

Nhiều toán tử phải nhớ trạng thái giữa các micro-batch — đó là stateful processing, lưu trong state store (mặc định backend HDFS-backed; có tùy chọn RocksDB state store cho state lớn để giảm áp lực JVM heap và GC):

• Aggregation có state — count, sum, avg... theo cửa sổ hoặc theo key. State là các giá trị trung gian đang tích luỹ.
• Stream–static join — nối một luồng với một bảng tĩnh (ví dụ làm giàu giao dịch bằng bảng khách hàng). Bảng tĩnh được đọc lại theo micro-batch; nhẹ về state.
• Stream–stream join — nối hai luồng. Vì cả hai đều vô tận, Spark phải buffer sự kiện của cả hai phía để chờ đối ứng, nên bắt buộc dùng watermark trên cả hai luồng cộng điều kiện ràng buộc theo thời gian để giới hạn buffer, nếu không state phình vô hạn.
• dropDuplicates — khử trùng lặp theo key; nên đi kèm watermark để không phải nhớ mọi key đã thấy mãi mãi (dropDuplicatesWithinWatermark giới hạn cửa sổ khử trùng theo watermark).

Điểm chung: state store là nơi sống còn của tính đúng đắn, và watermark là công cụ chính để giữ state hữu hạn.

Checkpoint và fault tolerance

Đây là phần làm nên độ tin cậy của Structured Streaming.

Khi khởi động query, bạn khai báo checkpointLocation trỏ tới một thư mục bền vững (HDFS, S3, ADLS...). Spark ghi vào đó:

• Offset log — với mỗi micro-batch, khoảng offset đã đọc từ nguồn (ví dụ offset Kafka theo partition). Đây là cách Spark biết "đã xử lý tới đâu".
• Commit log — đánh dấu batch nào đã hoàn tất.
• State store snapshots/deltas — trạng thái của các toán tử stateful.
• Metadata của query.

Khi query khởi động lại sau sự cố, nó đọc checkpoint để tiếp tục đúng chỗ đã dừng: đọc lại từ offset chưa commit và phục hồi state. Đây là nền tảng cho đảm bảo xử lý.

Exactly-once end-to-end

Structured Streaming đảm bảo exactly-once cho các phép tổng hợp khi kết hợp đủ ba điều kiện:

1. Nguồn phát lại được (replayable) theo offset — Kafka và file source thoả mãn (socket thì không).
2. State + offset lưu trong checkpoint — để sau crash tính lại từ đúng vị trí, không double-count.
3. Sink hỗ trợ ghi idempotent hoặc transactional — để cùng một dữ liệu ghi lại sau retry không tạo bản ghi trùng.

Nếu thiếu điều kiện thứ ba, bảo đảm thực tế xuống còn at-least-once: sau retry có thể có bản ghi trùng ở đích. Vì vậy khi ghi ra Kafka, ra Delta, hay dùng foreachBatch để ghi JDBC, hãy đảm bảo phía ghi là idempotent (ví dụ MERGE theo khóa, hoặc dùng batchId để khử trùng). Xem thêm cách sink transactional hoạt động trong bài Delta Lake & Lakehouse.

Một quy tắc vàng vận hành: mỗi query một checkpointLocation riêng, và đừng bao giờ xóa/chia sẻ nhầm nó — mất checkpoint đồng nghĩa mất offset và state, buộc phải xử lý lại và có thể sinh trùng.

Triggers

Trigger quyết định nhịp chạy micro-batch:

• Mặc định (không khai báo) — batch tiếp theo bắt đầu ngay khi batch trước xong; chạy liên tục hết công suất.
• processingTime="30 seconds" — chạy mỗi 30 giây một batch. Đều đặn, dễ kiểm soát tải.
• once=True (đã cũ) và availableNow=True (khuyến nghị thay thế) — xử lý toàn bộ dữ liệu đang có sẵn rồi dừng. Cực kỳ hữu ích: bạn viết logic streaming nhưng lịch chạy nó như một job batch định kỳ qua Airflow (xem series Airflow), tận dụng checkpoint để chỉ đọc phần mới kể từ lần trước. availableNow chia dữ liệu tồn thành nhiều batch nên an toàn hơn once khi tồn đọng lớn.
• continuous="1 second" — chế độ continuous thử nghiệm, checkpoint theo chu kỳ chỉ định.

availableNow là mẹo kiến trúc rất được ưa dùng: "streaming code, batch schedule" — vừa có ngữ nghĩa exactly-once và quản lý offset của streaming, vừa tiết kiệm chi phí vì không phải giữ cluster chạy 24/7.

Sơ đồ luồng

Ví dụ minh hoạ: windowed aggregation từ Kafka

# (minh hoạ) — đọc Kafka, tổng hợp theo cửa sổ event time có watermark, ghi có checkpoint
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, count, sum as _sum
from pyspark.sql.types import StructType, StringType, DoubleType, TimestampType

spark = SparkSession.builder.appName("txn-windowed-count").getOrCreate()

# Schema của message JSON trong Kafka
schema = (StructType()
          .add("txn_id", StringType())
          .add("account_id", StringType())
          .add("amount", DoubleType())
          .add("event_time", TimestampType()))

# 1) Đọc luồng từ Kafka
raw = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "broker:9092")
       .option("subscribe", "transactions")
       .option("startingOffsets", "latest")
       .load())

# 2) value là binary → CAST string → parse JSON thành các cột
txns = (raw
        .select(from_json(col("value").cast("string"), schema).alias("d"))
        .select("d.*"))

# 3) Watermark + tổng hợp theo cửa sổ tumbling 5 phút trên EVENT TIME
agg = (txns
       .withWatermark("event_time", "10 minutes")
       .groupBy(window(col("event_time"), "5 minutes"))
       .agg(count("*").alias("txn_count"),
            _sum("amount").alias("total_amount")))

# 4) Ghi ra sink với checkpoint (append mode: chốt sau khi cửa sổ qua watermark)
query = (agg
         .selectExpr("window.start AS win_start",
                     "window.end   AS win_end",
                     "txn_count", "total_amount")
         .writeStream
         .outputMode("append")
         .format("console")                # đổi thành kafka/delta/foreachBatch khi production
         .option("checkpointLocation", "/chk/txn-windowed-count")
         .trigger(processingTime="30 seconds")
         .start())

query.awaitTermination()

Muốn ghi vào PostgreSQL với ngữ nghĩa upsert, thay bước 4 bằng foreachBatch:

# (minh hoạ) — ghi từng micro-batch xuống JDBC một cách idempotent
def upsert_to_pg(batch_df, batch_id):
    # Ghi theo batch_id / khóa cửa sổ để retry không tạo bản ghi trùng
    (batch_df.write
        .format("jdbc")
        .option("url", "jdbc:postgresql://db:5432/warehouse")
        .option("dbtable", "txn_window_agg")
        .mode("append")            # thực tế nên MERGE theo (win_start, win_end)
        .save())

query = (agg.writeStream
         .outputMode("update")
         .foreachBatch(upsert_to_pg)
         .option("checkpointLocation", "/chk/txn-pg")
         .trigger(availableNow=True)   # xử lý phần tồn rồi dừng — chạy theo lịch Airflow
         .start())

Use case thực tế: đếm giao dịch theo cửa sổ 5 phút

Bài toán giám sát: đếm số giao dịch và tổng số tiền mỗi 5 phút theo thời điểm giao dịch thực tế, đổ về dashboard cảnh báo.

• Nguồn: topic Kafka transactions, mỗi message JSON có event_time do thiết bị ghi.
• Vì sao event time, không phải processing time? Vì thiết bị POS ở chi nhánh có thể mất mạng và đẩy trễ; nếu đếm theo lúc Spark đọc, một đợt giao dịch trễ dồn về sẽ làm phồng sai lệch cửa sổ hiện tại và làm rỗng cửa sổ quá khứ.
• Watermark 10 phút: chấp nhận giao dịch đến trễ tối đa 10 phút vẫn được tính vào đúng cửa sổ của nó; trễ hơn thì bỏ, và cửa sổ được chốt để giải phóng state.
• Window tumbling 5 phút: mỗi giao dịch rơi đúng một cửa sổ, cho con số đếm rõ ràng, không chồng lấn.
• Output mode: nếu dashboard cần con số cập nhật liên tục dùng update; nếu chỉ cần kết quả cuối cùng đã chốt ghi vào bảng append/lake thì dùng append.
• Đảm bảo: Kafka (replayable) + checkpoint (offset & state) + sink idempotent → exactly-once. Cảnh báo "cửa sổ nào vượt ngưỡng N giao dịch" từ đó là đáng tin, không double-count sau khi restart.

So sánh với cách làm cũ bằng Kafka Streams/ksqlDB trong bài Kafka Streams & ksqlDB: logic tương đương, nhưng Structured Streaming hợp hơn khi cùng team đã dùng Spark cho batch, muốn dùng lại code/skill và ghi thẳng vào lakehouse.

So với Flink — khi nào Structured Streaming là đủ

Apache Flink là engine streaming "native": xử lý từng sự kiện một (không chia micro-batch), nên có thể đạt độ trễ thấp hơn ổn định và có mô hình state/timer phong phú hơn cho các xử lý sự kiện phức tạp (CEP, state theo từng key rất tinh vi). Nếu bài toán đòi độ trễ hàng chục mili-giây bền vững, hoặc logic sự kiện rất phức tạp, Flink thường mạnh hơn.

Nhưng Structured Streaming đủ tốt cho phần lớn nhu cầu dữ liệu doanh nghiệp khi:

• Độ trễ giây/sub-giây là chấp nhận được (đúng với hầu hết dashboard, cảnh báo, ETL realtime).
• Bạn đã có hệ sinh thái Spark cho batch và muốn một engine, một API, một đội ngũ cho cả batch lẫn streaming.
• Đích đến là lakehouse (Delta/Iceberg) — nơi tích hợp Spark rất chặt.
• Bạn thích mẹo availableNow để chạy "streaming logic, batch cost".

Quy tắc thực dụng: bắt đầu bằng Structured Streaming vì chi phí học và vận hành thấp khi đã có Spark; chỉ chuyển sang Flink khi thực sự chạm giới hạn về độ trễ hoặc độ phức tạp state.

Ghi nhớ

• Mô hình unbounded table: luồng = bảng vô tận append liên tục; cùng DataFrame API như batch, chỉ đổi readStream/writeStream. Mặc định là micro-batch; continuous vẫn là thử nghiệm.
• Event time (lúc sự kiện xảy ra) mới có ý nghĩa nghiệp vụ, khác processing time (lúc Spark xử lý). Dữ liệu đến trễ là vấn đề trung tâm.
• Watermark (withWatermark) giới hạn thời gian chờ dữ liệu trễ: vừa quyết định khi nào chốt kết quả, vừa giải phóng state để khỏi phình vô hạn. Dài thì bắt nhiều trễ nhưng chậm/tốn; ngắn thì ngược lại.
• Window: tumbling (không chồng), sliding (chồng lấn), session (động theo hoạt động) — luôn tính trên event time và đi kèm watermark.
• Output mode: append (chỉ dòng đã bất biến — cần watermark với aggregation), update (dòng có thay đổi), complete (toàn bộ, chỉ cho kết quả nhỏ).
• Stateful: aggregation, stream–stream join (bắt buộc watermark hai phía), stream–static join, dropDuplicates; state để trong state store (có tùy chọn RocksDB).
• Checkpoint (checkpointLocation) lưu offset + state → khôi phục đúng chỗ. Exactly-once cần đủ: nguồn replayable + checkpoint + sink idempotent/transactional; thiếu sink idempotent thì chỉ at-least-once.
• Triggers: processingTime (nhịp đều), availableNow (xử lý phần tồn rồi dừng — "streaming code, batch schedule"), continuous (thử nghiệm).
• Chọn engine: Structured Streaming đủ cho hầu hết nhu cầu khi độ trễ giây là ổn và bạn đã có Spark; cân nhắc Flink khi cần độ trễ cực thấp hoặc xử lý sự kiện rất phức tạp.

Đọc tiếp

• Spark 3 — Spark SQL
• Spark 7 — Delta Lake & Lakehouse
• Kafka 1 — Kiến trúc
• Kafka 6 — Kafka Streams & ksqlDB