Spark 7 — Delta Lake & kiến trúc Lakehouse

Từ data lake "thô" đến Lakehouse

Trong bài Spark SQL và Structured Streaming, ta đã đọc/ghi Parquet trên một hệ thống file (HDFS, S3, ADLS, GCS). Cách tổ chức dữ liệu phổ biến nhất là thư mục Parquet có phân vùng: /sales/dt=2026-06-30/part-0001.parquet, part-0002.parquet,... Đây chính là data lake thô: rẻ, mở, chạy được với mọi engine. Nhưng khi đưa vào production, nó bộc lộ hàng loạt điểm yếu chí mạng.

• Không có ACID. Một job Spark ghi 200 file, chết ở file thứ 130. Giờ thư mục có 130 file "nửa vời". Người đọc song song có thể đọc trúng trạng thái dở dang. Không có "commit" nguyên tử — hoặc thấy hết, hoặc không thấy gì.
• Ghi dở làm hỏng bảng. Không có ranh giới giao dịch, việc dọn dẹp sau lỗi phải làm thủ công và dễ sai.
• Update/Delete rất khó. Parquet là bất biến (immutable). Muốn sửa 1 dòng phải đọc cả partition, sửa trong DataFrame, ghi đè lại toàn bộ file. Không có UPDATE/DELETE theo hàng đúng nghĩa — cực kỳ bất tiện cho GDPR "quyền được lãng quên" hay áp CDC.
• Không enforce schema. Ghi file có kiểu dữ liệu lệch (một job ghi amount là string, job khác là double) — không ai chặn. Đến lúc đọc mới vỡ.
• Đọc nhất quán yếu. "Đếm số file trong thư mục" không phải là một ảnh chụp (snapshot) nhất quán khi có người đang ghi.

Kho dữ liệu (data warehouse) truyền thống có tất cả những thứ này — ACID, transaction, ràng buộc schema — nhưng đóng, đắt, khó mở rộng cho dữ liệu bán cấu trúc và quy mô lớn. Lakehouse là ý tưởng ghép hai thế giới: giữ lưu trữ mở & rẻ của lake, thêm vào các đảm bảo giao dịch của warehouse. Lớp làm được điều đó gọi là open table format — và Delta Lake là một trong ba định dạng phổ biến nhất (cùng Apache Iceberg và Apache Hudi).

Delta Lake là gì?

Delta Lake là một định dạng bảng mở (open table format), xây trên file Parquet. Về vật lý, một bảng Delta vẫn là các file .parquet bình thường trong một thư mục. Điều Delta thêm vào là một thư mục con đặc biệt: _delta_log — transaction log (còn gọi là DeltaLog).

_delta_log là một chuỗi bản ghi có thứ tự của mọi giao dịch đã commit lên bảng kể từ khi tạo. Mỗi commit là một file JSON được đánh số tăng dần:

mytable/
├── _delta_log/
│   ├── 00000000000000000000.json   ← commit v0
│   ├── 00000000000000000001.json   ← commit v1
│   ├── 00000000000000000002.json   ← commit v2
│   └── 00000000000000000010.checkpoint.parquet  ← checkpoint mỗi 10 commit
├── part-00000-....parquet
├── part-00001-....parquet
└── ...

Mỗi file JSON ghi các action: "thêm file X (kèm thống kê min/max)", "gỡ file Y", "đổi metadata/schema", "commit info". Trạng thái hiện tại của bảng = phát lại (replay) toàn bộ log từ v0 tới bản mới nhất: file nào add mà chưa remove thì đang thuộc bảng. Để không phải đọc hàng nghìn file JSON, Delta ghi checkpoint (mặc định mỗi 10 commit) gộp trạng thái vào một file Parquet, rồi chỉ đọc checkpoint + các JSON sau đó.

Cơ chế log này mang lại các đảm bảo cốt lõi:

• Atomic commit (ACID). Job ghi 200 file nhưng bảng chỉ thay đổi khi file JSON commit được ghi thành công (một thao tác nguyên tử). Nếu job chết giữa chừng, không có file JSON mới → người đọc vẫn thấy phiên bản cũ nguyên vẹn; các file Parquet mồ côi đơn giản bị bỏ qua (và được VACUUM dọn sau).
• Đọc nhất quán (snapshot isolation). Người đọc chốt một phiên bản log tại thời điểm bắt đầu và đọc đúng ảnh chụp đó, dù có ai đang ghi.
• Ghi đồng thời (optimistic concurrency). Nhiều writer cùng ghi: mỗi writer đọc phiên bản hiện tại, làm việc, rồi cố commit ở phiên bản kế tiếp. Nếu phiên bản đó đã bị người khác chiếm, Delta phát hiện xung đột và (tuỳ loại) thử lại hoặc báo lỗi — không "ghi đè mù".

Các tính năng chính

Time travel (đi ngược thời gian)

Vì log lưu toàn bộ lịch sử, ta có thể tái tạo trạng thái bảng ở bất kỳ phiên bản nào — gọi là time travel hay data versioning. Rất hữu ích để audit, debug pipeline, hoặc "hoàn tác" một lần ghi sai.

-- (minh hoạ) — Delta SQL
-- Đọc theo số phiên bản
SELECT * FROM sales VERSION AS OF 42;

-- Đọc theo mốc thời gian
SELECT * FROM sales TIMESTAMP AS OF '2026-06-28 09:00:00';

-- So sánh: có bao nhiêu dòng bị thay đổi giữa 2 phiên bản?
SELECT count(*) FROM sales VERSION AS OF 43
EXCEPT
SELECT count(*) FROM sales VERSION AS OF 42;

# (minh hoạ) — PySpark
df_before = (spark.read.format("delta")
             .option("versionAsOf", 42)
             .load("/lake/gold/sales"))

Schema enforcement & schema evolution

Mặc định Delta enforce schema: ghi một DataFrame có cột kiểu lệch với bảng sẽ bị từ chối ngay tại thời điểm ghi (fail-fast), tránh dữ liệu bẩn lọt vào. Khi cố ý muốn thêm cột mới, bật schema evolution:

# (minh hoạ)
(df.write.format("delta")
   .mode("append")
   .option("mergeSchema", "true")   # cho phép thêm cột mới
   .save("/lake/silver/customer"))

MERGE — upsert cho CDC/SCD

MERGE là tính năng mạnh nhất cho kỹ sư dữ liệu: upsert (update nếu khớp, insert nếu không) theo một điều kiện — làm được với dữ liệu lake dễ như UPDATE trên CSDL quan hệ. Đây là cách chuẩn để áp CDC hoặc dựng SCD (Slowly Changing Dimension).

-- (minh hoạ) — áp một batch thay đổi CDC vào bảng silver
MERGE INTO silver.customer AS t
USING staging_cdc AS s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED AND s.op IN ('U','I') THEN UPDATE SET *
WHEN NOT MATCHED AND s.op IN ('U','I') THEN INSERT *;

Delta cũng có DELETE/UPDATE theo hàng đúng nghĩa (khác hẳn data lake thô):

-- (minh hoạ)
DELETE FROM silver.customer WHERE gdpr_erasure = true;
UPDATE silver.customer SET status = 'inactive' WHERE last_seen < '2025-01-01';

OPTIMIZE, Z-ORDER và VACUUM

Vấn đề file nhỏ (small files). Streaming và micro-batch tạo ra hàng loạt file bé; đọc chúng rất chậm vì overhead mở file. OPTIMIZE (compaction) gộp nhiều file nhỏ thành ít file lớn hơn.

ZORDER BY sắp xếp lại dữ liệu theo một/nhiều cột để các giá trị gần nhau nằm cùng file — nhờ đó data skipping hiệu quả hơn. Delta lưu thống kê min/max từng file trong log; khi query có WHERE, engine bỏ qua (skip) các file mà khoảng min/max không thể chứa kết quả. Z-ORDER phát huy tốt nhất với cột có độ phân biệt cao (high cardinality) thường dùng trong bộ lọc.

-- (minh hoạ)
OPTIMIZE silver.transactions
  WHERE dt >= '2026-06-01'
  ZORDER BY (account_id, txn_ts);

Từ Delta Lake 3.1 trở đi còn có Auto Compaction — tự gộp file nhỏ ngay sau khi ghi thành công, giảm nhu cầu chạy OPTIMIZE thủ công.

VACUUM dọn các file Parquet không còn được phiên bản nào tham chiếu (file bị remove sau OPTIMIZE, DELETE, hoặc job lỗi). Lưu ý: VACUUM có ngưỡng lưu giữ (mặc định 7 ngày) — dọn quá sâu sẽ phá vỡ time travel về các phiên bản cũ.

-- (minh hoạ) — giữ lại 168 giờ (7 ngày) file cũ để còn time travel
VACUUM silver.transactions RETAIN 168 HOURS;

Kiến trúc Medallion: Bronze → Silver → Gold

Có bảng giao dịch đáng tin rồi, câu hỏi tiếp theo là tổ chức nhiều bảng thành một pipeline mạch lạc. Mẫu phổ biến nhất trên lakehouse là Medallion architecture — nâng dần chất lượng dữ liệu qua ba lớp.

• Bronze (raw). Nơi hạ cánh dữ liệu thô từ nguồn, gần như nguyên trạng, thường append-only kèm metadata nạp (nguồn, thời điểm, offset). Mục tiêu: có bản gốc bất biến để nạp lại được. Đây là điểm tiếp nhận các message CDC/Kafka.
• Silver (cleaned/conformed). Dữ liệu Bronze được làm sạch, chuẩn hoá kiểu, khử trùng lặp, hợp nhất (conform) thành "góc nhìn doanh nghiệp" của các thực thể chính (khách hàng, giao dịch, tài khoản). Đây là nơi MERGE áp CDC để giữ trạng thái hiện tại từng bản ghi.
• Gold (aggregated/serving). Bảng tổng hợp phục vụ tiêu dùng: fact/dimension theo mô hình chiều, KPI, bảng phục vụ báo cáo và ML. Đây là bề mặt cho BI.

Chi tiết mô hình hoá và governance ở các lớp này được bàn kỹ trong bài mô hình kho dữ liệu và bài quản trị/chất lượng của series Data Engineering. Điểm mấu chốt: mỗi lớp là bảng Delta — nên mỗi bước biến đổi đều có ACID, có time travel, có thể chạy lại an toàn.

Đưa CDC vào Lakehouse

Đây là nơi Delta ăn khớp với cả series. Chuỗi CDC (Oracle CDC) và Kafka Connect (Kafka Connect) tạo ra dòng thay đổi (INSERT/UPDATE/DELETE) từ nguồn OLTP. Luồng điển hình:

1. Debezium/GoldenGate đọc redo log Oracle → phát message change lên Kafka.
2. Spark Structured Streaming (hoặc Kafka Connect sink) đọc topic, ghi thô vào Bronze (append-only, giữ cả trước/sau và cờ thao tác op).
3. Micro-batch dùng MERGE áp change từ Bronze vào Silver để phản chiếu trạng thái nguồn (upsert theo khoá, xoá theo op='D').
4. Job tổng hợp cập nhật Gold phục vụ báo cáo.

# (minh hoạ) — foreachBatch: áp CDC vào Silver bằng MERGE trong streaming
from delta.tables import DeltaTable

def upsert_to_silver(microbatch_df, batch_id):
    # Với mỗi khoá, chỉ giữ bản ghi CDC mới nhất trong batch
    from pyspark.sql import functions as F
    from pyspark.sql.window import Window
    w = Window.partitionBy("customer_id").orderBy(F.col("source_ts").desc())
    latest = (microbatch_df
              .withColumn("rn", F.row_number().over(w))
              .filter("rn = 1").drop("rn"))

    target = DeltaTable.forName(spark, "silver.customer")
    (target.alias("t")
       .merge(latest.alias("s"), "t.customer_id = s.customer_id")
       .whenMatchedDelete(condition="s.op = 'D'")
       .whenMatchedUpdateAll(condition="s.op in ('U','I')")
       .whenNotMatchedInsertAll(condition="s.op in ('U','I')")
       .execute())

(spark.readStream.format("delta").table("bronze.customer_cdc")
     .writeStream
     .foreachBatch(upsert_to_silver)
     .option("checkpointLocation", "/lake/_chk/silver_customer")
     .trigger(availableNow=True)   # xử lý hết dữ liệu sẵn có rồi dừng
     .start())

Lưu ý thực chiến: de-dup trong từng micro-batch trước khi MERGE (như trên) để tránh nhiều thay đổi cùng khoá va nhau; và MERGE của Delta còn hỗ trợ Change Data Feed (CDF) để chính bảng Delta phát ra dòng thay đổi cho lớp sau tiêu thụ.

Use case thực tế

1) Áp CDC vào bảng silver khách hàng (ngân hàng). Core Banking chạy Oracle. GoldenGate/Debezium đẩy thay đổi bảng CUSTOMER lên Kafka. Bronze giữ toàn bộ event thô (audit đầy đủ). Job streaming foreachBatch gọi MERGE vào silver.customer: cập nhật hồ sơ, xoá khi khách bị đóng. Kết quả: bảng silver luôn phản chiếu nguồn với độ trễ vài phút, mà không cần full-load hằng đêm.

2) Time travel để audit & khôi phục. Một sáng, báo cáo gold.daily_balance nhảy số bất thường. Nhờ time travel, phân tích viên chạy SELECT ... VERSION AS OF <n-1> để so trạng thái trước/sau lần ghi nghi vấn, khoanh vùng đúng batch gây lỗi. Nếu do một lần MERGE sai, có thể RESTORE TABLE ... TO VERSION AS OF để khôi phục về phiên bản tốt — đảo ngược sự cố mà không cần backup ngoài.

-- (minh hoạ)
DESCRIBE HISTORY gold.daily_balance;          -- xem lịch sử commit
RESTORE TABLE gold.daily_balance TO VERSION AS OF 118;

Delta vs Apache Iceberg vs Apache Hudi

Cả ba đều là open table format giải quyết cùng bài toán: ACID + time travel + schema evolution + upsert trên object store. Khác biệt nằm ở nguồn gốc và điểm mạnh:

Xu hướng 2025: ngành hội tụ mạnh về chuẩn mở, và Iceberg nổi lên như tiêu chuẩn tương tác (interop) được nhiều cloud hỗ trợ native. Đáp lại, Delta đưa ra UniForm (Universal Format): một bảng Delta có thể đồng thời sinh metadata Iceberg (và Hudi) để engine của các format khác đọc được — dù thao tác ghi vẫn đi qua giao thức Delta (metadata Iceberg là read-only). Ngoài ra, Apache XTable (trước là OneTable) dịch metadata đa chiều giữa ba format mà không copy dữ liệu. Thông điệp: chọn format không còn là quyết định "một chiều không thể đảo" như trước.

Vai trò trong nền dữ liệu hiện đại

Delta/lakehouse là tầng lưu trữ giao dịch gắn kết các mảnh của series lại:

• Compute: Spark đọc/ghi Delta; Structured Streaming ghi micro-batch vào Bronze/Silver.
• Ingest: Oracle CDC + Kafka Connect đưa thay đổi vào lake; MERGE áp vào Silver.
• Điều phối: Airflow lập lịch các bước bronze→silver→gold và OPTIMIZE/VACUUM định kỳ.
• Serving/BI: Gold là bề mặt cho BI. Với hệ Microsoft Fabric/Power BI, Direct Lake cho phép Power BI đọc trực tiếp file Delta ở lớp Gold — không import, không cache trung gian — kết hợp tốc độ import với độ tươi của DirectQuery.

Ghi nhớ

• Delta = Parquet + transaction log (_delta_log). Log là nguồn sự thật; trạng thái bảng = replay log; checkpoint để đọc nhanh.
• ACID nhờ atomic commit + optimistic concurrency: job chết giữa chừng không làm hỏng bảng; đọc luôn thấy snapshot nhất quán.
• MERGE là công cụ vàng để áp CDC/SCD — upsert theo khoá, xoá theo op='D'; nhớ de-dup trong batch trước khi merge.
• Time travel cho audit và RESTORE để khôi phục nhanh; nhưng VACUUM cắt file cũ sẽ giới hạn time travel — cân nhắc ngưỡng lưu giữ.
• OPTIMIZE + ZORDER trị file nhỏ và tăng data skipping; Z-ORDER hợp cột high-cardinality trong bộ lọc.
• Medallion (Bronze→Silver→Gold) là mẫu tổ chức chuẩn; mỗi lớp là bảng Delta nên đều có ACID và chạy lại an toàn.
• Delta/Iceberg/Hudi cùng giải một bài toán; xu hướng hội tụ mở (UniForm, XTable) khiến lựa chọn bớt "khoá cứng".

Tham khảo: Delta Lake — Transaction Log (Databricks) · Delta Lake Optimizations (docs) · What is Medallion Architecture (Databricks) · Building the Medallion Architecture (delta.io) · Iceberg vs Delta vs Hudi (Onehouse)