Dựng pipeline batch bằng Spark và stream bằng Kafka, Flink

Khi quy mô đủ lớn, bạn sẽ cần đến Spark, Airflow, Kafka và Flink. Bài này đi qua các bước dựng cả hai loại pipeline.

Code trong bài này là mẫu để đọc và điều chỉnh, chưa được chạy trên một cluster thật. Hãy thử trên máy cá nhân trước.

Dựng pipeline batch gồm những bước nào?#

1. Chốt yêu cầu. Ai dùng kết quả, cần mới đến mức nào, dữ liệu nằm ở đâu.
2. Đưa dữ liệu vào lake. Sao chép vào object storage theo định dạng cột như Parquet, chia thư mục theo ngày.
3. Viết job biến đổi. Mỗi lần chạy xử lý một khoảng thời gian rõ ràng, ví dụ một ngày.
4. Làm cho job idempotent. Chạy lại cùng một ngày phải cho cùng kết quả. Ghi đè đúng phân vùng của ngày đó.
5. Điều phối bằng Airflow. Khai báo lịch, thứ tự, số lần thử lại, cảnh báo khi lỗi.
6. Giám sát chất lượng. Đo thời gian chạy, số dòng vào ra, kiểm tra null, trùng, lệch tổng. Chuẩn bị sẵn cách backfill.

# daily_revenue.py: tính doanh thu theo cửa hàng cho một ngày
import sys
from pyspark.sql import SparkSession, functions as F

day = sys.argv[1]  # ví dụ 2026-10-25
spark = (SparkSession.builder.appName("daily_revenue")
         .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
         .getOrCreate())

orders = spark.read.parquet(f"s3a://lake/orders/dt={day}/")

daily = (orders
    .where(F.col("status") == "PAID")
    .groupBy("shop_id")
    .agg(F.count("*").alias("so_don"),
         F.sum("amount").alias("doanh_thu")))

# Idempotent: chỉ ghi đè phân vùng của ngày này
(daily.withColumn("dt", F.lit(day))
      .write.mode("overwrite")
      .partitionBy("dt")
      .parquet("s3a://warehouse/daily_revenue/"))

Thiếu partitionOverwriteMode=dynamic, lệnh ghi đè sẽ xoá sạch mọi phân vùng khác.

# dags/daily_revenue.py: Airflow 3
from datetime import datetime
from airflow.sdk import dag, task

# Chạy 02:00 mỗi đêm, tự bù các ngày bị bỏ lỡ (catchup)
@dag(schedule="0 2 * * *", start_date=datetime(2026, 10, 1), catchup=True)
def daily_revenue():

    @task.bash(retries=2)
    def run_spark():
        # {{ ds }} là ngày logic của lần chạy, giúp backfill đúng ngày
        return "spark-submit daily_revenue.py {{ ds }}"

    run_spark()

daily_revenue()

Dựng pipeline stream gồm những bước nào?#

1. Xác định sự kiện. Gồm id, event time và dữ liệu. Chọn key để chia partition, ví dụ shop_id.
2. Đưa sự kiện vào Kafka. Chọn số partition theo thông lượng cần, vì partition quyết định mức song song.
3. Chọn engine. Flink khi cần trạng thái lớn và event time chính xác. Kafka Streams khi logic nhỏ nằm trong một dịch vụ Java. Spark khi đội đã quen Spark.
4. Khai báo thời gian và cửa sổ. Đặt event time, watermark và loại cửa sổ.
5. Bật checkpoint và chọn đích. Ghi ra đích kiểu upsert theo khoá để không nhân đôi khi xử lý lại.
6. Vận hành. Theo dõi consumer lag, thời gian checkpoint và backpressure. Thử tắt một worker để xem hệ thống phục hồi thế nào.

# producer.py: gửi đơn hàng vào topic "orders"
import json, time
from confluent_kafka import Producer

p = Producer({"bootstrap.servers": "localhost:9092"})

def send_order(order_id, shop_id, amount):
    event = {"order_id": order_id, "shop_id": shop_id,
             "amount": amount,
             "ts": int(time.time() * 1000)}  # event time, mili giây
    # key = shop_id: đơn cùng cửa hàng vào cùng partition
    p.produce("orders", key=shop_id, value=json.dumps(event))

send_order("A1021", "shop-7", 249000)
p.flush()

-- 1) Khai báo nguồn, cho phép sự kiện trễ tối đa 5 giây
CREATE TABLE orders (
  order_id STRING,
  shop_id  STRING,
  amount   BIGINT,
  ts       BIGINT,
  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'localhost:9092',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);

-- 2) Đếm đơn và doanh thu mỗi cửa hàng theo cửa sổ 1 phút
SELECT window_start, window_end, shop_id,
       COUNT(*) AS so_don, SUM(amount) AS doanh_thu
FROM TABLE(TUMBLE(TABLE orders, DESCRIPTOR(ts_ltz), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, shop_id;

Epoch mili giây không đọc được trực tiếp thành TIMESTAMP_LTZ, nên mẫu giữ ts BIGINT rồi chuyển đổi.

Nên học tiếp theo thứ tự nào?#

Hai bài tập để thực hành:

• Bài tập batch. Dùng PySpark trên máy cá nhân để tính doanh thu theo ngày từ một file CSV đơn hàng.
• Bài tập stream. Dựng Kafka và Flink bằng Docker Compose, viết producer giả lập đơn hàng và đếm số đơn mỗi phút bằng Flink SQL.

Sai lầm hay gặp#

• Nối thêm thay vì ghi đè phân vùng. Mỗi lần chạy lại sẽ nhân đôi số liệu.
• Quên catchup và ngày logic. Backfill sẽ tính sai ngày nếu job dùng giờ hiện tại thay cho {{ ds }}.
• Chọn key lệch. Một cửa hàng quá lớn dồn hết vào một partition và làm nghẽn cả hệ thống.
• Không thử tắt máy. Chỉ khi tắt thử một worker bạn mới biết checkpoint có phục hồi đúng hay không.