Databricks 4 — Workflows & Delta Live Tables

Từ compute đến điều phối

Databricks 3 — Compute trả lời câu hỏi chạy code ở đâu. Bài này trả lời câu hỏi kế tiếp: chạy code gì, theo thứ tự nào, khi nào, và làm sao biết nó chạy đúng. Đó là bài toán điều phối (orchestration) và xây dựng pipeline.

Databricks đưa ra hai công cụ bổ sung cho nhau:

• Databricks Workflows — bộ điều phối gốc, chạy các task bất kỳ (notebook, Python, JAR, SQL, DLT...) theo một DAG phụ thuộc, có lịch/trigger/retry/cảnh báo. Đây là lớp quản lý luồng công việc.
• Delta Live Tables (DLT) — framework khai báo để xây pipeline ETL: bạn định nghĩa các bảng đầu ra, DLT tự lo dependency, thứ tự, incremental, checkpoint, retry và data quality. Đây là lớp transformation dữ liệu.

Quan hệ giữa hai lớp: một pipeline DLT thường được kích hoạt như một task bên trong một Workflows job. Workflows lo lịch và ghép nối; DLT lo phần biến đổi dữ liệu và chất lượng.

Databricks Workflows

Workflows là công cụ điều phối tích hợp sẵn của Databricks. Đơn vị cơ bản là một job; mỗi job gồm một hoặc nhiều task, và các task nối với nhau thành một DAG (Directed Acyclic Graph — đồ thị có hướng không chu trình) qua quan hệ phụ thuộc.

Task — các loại việc chạy được

Một task là một đơn vị công việc. Databricks hỗ trợ nhiều loại task:

Việc trộn được nhiều loại task trong một job là điểm mạnh: ví dụ ingest bằng notebook → transform bằng DLT → chạy kiểm thử bằng dbt → refresh dashboard bằng task SQL, tất cả trong một DAG.

DAG — phụ thuộc giữa task

Mỗi task khai báo nó phụ thuộc task nào (depends_on). Databricks tự dựng DAG, chạy song song các nhánh độc lập và tuần tự các nhánh phụ thuộc. Có thể đặt điều kiện chạy (run if): chạy task tiếp theo chỉ khi các task trước "all succeeded", "at least one failed", "all done"... — dùng để rẽ nhánh xử lý lỗi.

Trigger — khi nào job chạy

• Scheduled (cron) — chạy theo lịch định kỳ (biểu thức cron/quartz, kèm múi giờ). Ví dụ ETL đêm 02:00.
• File arrival — chạy khi có file mới xuất hiện tại một vị trí cloud storage (event-driven). Hợp cho nạp file đối tác gửi bất định giờ.
• Continuous — job chạy liên tục: vừa xong lượt này lại bắt lượt sau ngay, dùng cho xử lý gần thời gian thực.
• Manual / API / trigger từ job khác — chạy theo yêu cầu hoặc từ hệ thống ngoài qua REST API.

Retry, cảnh báo, tham số, job cluster

• Retry: cấu hình số lần thử lại và khoảng chờ theo từng task — task chạm lỗi tạm thời (mất mạng, spot bị thu hồi) tự thử lại thay vì fail cả job.
• Cảnh báo (alerts/notifications): gửi email/webhook/Slack/PagerDuty khi job start, success, failure, hoặc chạy quá lâu (duration threshold) — bắt kịp job "treo".
• Tham số (parameters): truyền tham số vào job (ví dụ ngày xử lý run_date), đọc trong task qua widget/dbutils. Cho phép backfill cùng một job cho nhiều ngày.
• Job cluster: mỗi task/job có thể chỉ định một job cluster (tạo khi chạy, huỷ khi xong) — rẻ và sạch hơn all-purpose (xem Databricks 3). Nhiều task có thể dùng chung một job cluster để tiết kiệm thời gian khởi động.

Workflows so với Airflow

Databricks Workflows và Apache Airflow đều là orchestrator, nhưng phạm vi khác nhau:

Nguyên tắc chọn: nếu pipeline gần như toàn bộ nằm trong Databricks, dùng Workflows tích hợp sẵn — đơn giản, không phải nuôi một cụm Airflow. Nếu bạn phải điều phối nhiều hệ thống hỗn hợp (kích Databricks và gọi API bên thứ ba, đẩy file, chạy job on-prem trong cùng một luồng có SLA chung), Airflow (hoặc một orchestrator trung lập) là lựa chọn hợp lý — khi đó Airflow thường gọi Databricks job qua operator, không thay thế phần transform bên trong.

Delta Live Tables (DLT) — pipeline khai báo

Với ETL truyền thống trên Spark, bạn viết code mệnh lệnh (imperative): tự đọc nguồn, tự sắp thứ tự bảng, tự quản checkpoint streaming, tự viết logic incremental, tự xử lý retry. Dễ sai và khó bảo trì khi pipeline lớn.

Delta Live Tables (DLT) đảo ngược cách nghĩ: bạn viết khai báo (declarative) — định nghĩa các bảng đầu ra bằng SQL hoặc Python và mối quan hệ giữa chúng, rồi DLT tự lo phần còn lại:

• Suy ra dependency & thứ tự chạy từ các tham chiếu bảng (ai đọc ai) → tự dựng DAG dữ liệu.
• Quản lý incremental & checkpoint cho streaming — bạn không tự viết offset/checkpoint.
• Retry, phục hồi lỗi, dừng gọn khi hỏng.
• Data quality inline qua expectations (mục dưới).
• Event log & metrics cho mỗi lần chạy, phục vụ giám sát/lineage.

Bạn khai báo pipeline; DLT là runtime biến khai báo đó thành một pipeline Delta chạy được.

Streaming table vs materialized view

DLT có hai kiểu đối tượng chính, chọn đúng là mấu chốt:

Kinh nghiệm: streaming table cho lớp nạp và làm sạch từng dòng (nơi dữ liệu chảy vào liên tục và bạn không muốn quét lại toàn bảng), materialized view cho lớp tổng hợp (nơi cần kết quả luôn đúng với logic mới nhất).

Ví dụ DLT bằng SQL

Ví dụ minh hoạ (KHÔNG đánh dấu chạy được — đây là cú pháp DLT trên Databricks, không phải PostgreSQL sandbox):

-- Bronze: nạp giao dịch thô từ cloud storage (streaming)
CREATE OR REFRESH STREAMING TABLE txn_bronze
COMMENT "Giao dịch thô từ core banking, nạp bằng Auto Loader"
AS SELECT *, current_timestamp() AS ingest_ts
FROM STREAM read_files('/mnt/landing/txn/', format => 'json');

-- Silver: làm sạch + kiểm chất lượng inline
CREATE OR REFRESH STREAMING TABLE txn_silver (
  CONSTRAINT valid_amount   EXPECT (amount IS NOT NULL AND amount <> 0) ON VIOLATION DROP ROW,
  CONSTRAINT valid_account  EXPECT (account_id IS NOT NULL),
  CONSTRAINT valid_currency EXPECT (currency IN ('VND','USD','EUR')) ON VIOLATION FAIL UPDATE
)
AS SELECT account_id, amount, currency, kind,
          CAST(created_at AS TIMESTAMP) AS created_at
FROM STREAM(LIVE.txn_bronze);

-- Gold: tổng hợp phục vụ báo cáo (materialized view)
CREATE OR REFRESH MATERIALIZED VIEW daily_account_txn
AS SELECT account_id, DATE(created_at) AS d,
          SUM(amount) AS total, COUNT(*) AS n_txn
FROM LIVE.txn_silver
GROUP BY account_id, DATE(created_at);

Chú ý: LIVE.<tên> (hoặc tên trong cùng pipeline) là cách một bảng DLT tham chiếu bảng DLT khác — chính từ các tham chiếu này DLT suy ra DAG. STREAM(...) báo đọc incremental.

Ví dụ DLT bằng Python

import dlt
from pyspark.sql.functions import col, current_timestamp

@dlt.table(comment="Giao dịch thô")
def txn_bronze():
    return (spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/mnt/landing/txn/")
            .withColumn("ingest_ts", current_timestamp()))

@dlt.table(comment="Giao dịch đã làm sạch")
@dlt.expect_or_drop("valid_amount", "amount IS NOT NULL AND amount <> 0")
@dlt.expect("valid_account", "account_id IS NOT NULL")
def txn_silver():
    return (dlt.read_stream("txn_bronze")
            .select("account_id", "amount", "currency", "kind",
                    col("created_at").cast("timestamp").alias("created_at")))

Expectations — kiểm chất lượng inline

Expectations là ràng buộc chất lượng dữ liệu (data quality) khai báo ngay trong định nghĩa bảng DLT — thay vì kiểm chất lượng ở một bước tách rời sau pipeline. Cú pháp SQL: CONSTRAINT <tên> EXPECT (<điều kiện>) ON VIOLATION <hành động>.

Ba hành động khi dòng vi phạm điều kiện:

Điểm mạnh: mỗi expectation phát ra metrics (số dòng pass/fail theo từng rule) vào event log của pipeline — bạn có một bảng điểm chất lượng theo thời gian, không cần tự dựng. Đây là data quality gắn liền pipeline, bổ sung cho cách kiểm chất lượng ở tầng quản trị dữ liệu (Governance 3 — Data Quality).

Nguyên tắc thực chiến: drop cho lỗi dữ liệu "bẩn nhưng chấp nhận được" (thiếu field không quan trọng), fail cho ràng buộc sống-còn (số tiền âm bất thường, mã tiền tệ lạ — có thể là lỗi nguồn nghiêm trọng), warn cho chỉ báo cần theo dõi mà chưa chặn.

Auto Loader — nạp file incremental

Auto Loader là cơ chế nạp file incremental hiệu quả từ cloud storage (S3/ADLS/GCS). Thay vì mỗi lần liệt kê lại toàn bộ thư mục (chậm và tốn khi có hàng triệu file), Auto Loader theo dõi file nào đã xử lý và chỉ nạp file mới. Nó dùng nguồn cloudFiles trong Structured Streaming (và là cách chuẩn để nạp vào streaming table của DLT như ví dụ trên).

Hai tính năng quan trọng:

• Schema inference — tự suy ra schema từ dữ liệu, không cần khai báo tay từng cột.
• Schema evolution — khi nguồn thêm cột mới, Auto Loader phát hiện và cho phép tiến hoá schema (thêm cột vào bảng) thay vì gãy pipeline. Có thể cấu hình chính sách xử lý cột lạ (thêm, đưa vào cột "rescued", hay fail).

Cơ chế phát hiện file mới: mặc định directory listing (liệt kê tối ưu), hoặc file notification (đăng ký sự kiện từ cloud để biết ngay khi có file, cực hiệu quả với thư mục cực lớn). Trong ngân hàng, Auto Loader hợp với nạp file sao kê/đối soát đối tác gửi định kỳ vào lớp bronze.

Medallion bằng DLT

Kiến trúc medallion chia dữ liệu thành ba lớp chất lượng tăng dần — bronze → silver → gold — và DLT là công cụ hiện thực nó tự nhiên nhất (mỗi lớp là một tập bảng DLT):

• Bronze: bản sao thô của nguồn, ít biến đổi nhất — chỉ nạp (thường bằng Auto Loader) và gắn metadata (thời điểm nạp, tên file). Giữ nguyên để có thể phát lại (replay).
• Silver: làm sạch, chuẩn hoá kiểu, khử trùng lặp, áp expectations. Dữ liệu ở đây đã "tin được" cho phân tích.
• Gold: tổng hợp/join thành bảng phục vụ trực tiếp cho BI, báo cáo, hoặc feature ML — thường là materialized view.

Về nền tảng lakehouse & medallion nói chung, xem Data Engineering 7 — Lakehouse và Databricks 2 — Delta Lake.

CDC với APPLY CHANGES INTO

Nhiều nguồn ngân hàng không phải append-only mà là CDC (Change Data Capture) — dòng dữ liệu gồm insert/update/delete từ hệ thống nguồn (ví dụ core banking qua Debezium/GoldenGate). Áp thủ công CDC vào bảng đích (merge đúng thứ tự, xử lý update đến trước insert...) rất dễ sai.

DLT cung cấp APPLY CHANGES INTO — xử lý CDC khai báo:

-- Minh hoạ (không đánh dấu chạy được): áp CDC vào bảng khách hàng
CREATE OR REFRESH STREAMING TABLE customers_silver;

APPLY CHANGES INTO LIVE.customers_silver
FROM STREAM(LIVE.customers_cdc)
KEYS (customer_id)
APPLY AS DELETE WHEN op = 'D'
SEQUENCE BY commit_ts          -- sắp thứ tự thay đổi theo thời điểm commit
COLUMNS * EXCEPT (op, commit_ts)
STORED AS SCD TYPE 1;          -- hoặc SCD TYPE 2 để giữ lịch sử

DLT tự lo: khử trùng lặp, sắp đúng thứ tự thay đổi theo cột SEQUENCE BY (dù dữ liệu đến lệch giờ), xử lý delete, và chọn SCD Type 1 (ghi đè) hay Type 2 (giữ lịch sử phiên bản — xem thêm mô hình SCD ở Data Modeling). Đây chính là chỗ DLT tiết kiệm nhiều công sức nhất so với tự viết merge.

Giám sát pipeline

DLT ghi một event log có cấu trúc cho mỗi pipeline — một bảng Delta bạn truy vấn được. Trong đó có:

• Trạng thái từng lần chạy & từng bảng: bắt đầu/hoàn thành, số dòng đọc/ghi, thời gian.
• Data quality metrics: với mỗi expectation, số dòng pass/fail — dựng dashboard "sức khoẻ dữ liệu" theo thời gian.
• Lineage: quan hệ bảng→bảng mà DLT suy ra, hiển thị đồ hoạ và đẩy vào Unity Catalog lineage (Governance 2 — Catalog & Lineage).
• Sự cố: lỗi khiến pipeline dừng, để điều tra.

Workflows bổ sung lớp giám sát điều phối: lịch sử run của job, thời lượng, task nào fail, cảnh báo khi vượt ngưỡng thời gian. Kết hợp lại: Workflows cho biết job có chạy đúng lịch và đúng luồng không; DLT event log cho biết dữ liệu bên trong có sạch không.

Use case thực tế

Bối cảnh (NCB): Xây pipeline ELT giao dịch chạy đêm, nạp ~40 triệu giao dịch/ngày từ file JSON core banking đổ vào cloud storage, biến đổi qua medallion, có kiểm chất lượng tự động, phục vụ báo cáo giám sát giao dịch.

Thiết kế:

1. DLT pipeline (medallion):

   • Bronze txn_bronze — streaming table nạp bằng Auto Loader (file notification, schema evolution bật để chịu được nguồn thêm cột).
   • Silver txn_silver — streaming table với expectations: EXPECT amount IS NOT NULL ON VIOLATION DROP (bỏ dòng rác), EXPECT currency IN ('VND','USD','EUR') ON VIOLATION FAIL (mã tiền lạ = lỗi nguồn nghiêm trọng, dừng pipeline để điều tra).
   • Gold daily_account_txn — materialized view tổng hợp theo tài khoản/ngày.
   • Bảng chiều khách hàng nạp qua APPLY CHANGES INTO ... SCD TYPE 2 để giữ lịch sử thay đổi thông tin KH (phục vụ đối chiếu thời điểm).

2. Workflows job điều phối:

   • Trigger cron 02:00 (múi giờ Asia/Ho_Chi_Minh).
   • Task 1: notebook kiểm tra file nguồn đã đủ. Task 2: DLT pipeline (phụ thuộc task 1). Task 3: task SQL refresh dashboard giám sát (phụ thuộc task 2, run if all succeeded).
   • Job cluster với autoscale + worker spot (job có retry).
   • Retry 3 lần cho task nạp; alert Slack khi fail hoặc job chạy quá 90 phút.
   • Tham số run_date để backfill khi cần nạp lại một ngày cũ.

Kết quả minh hoạ: Vi phạm chất lượng được đếm tự động mỗi đêm (ví dụ 0,3% dòng bị drop do thiếu amount) và hiện lên dashboard từ event log; một đêm nguồn lỗi phát sinh mã tiền tệ lạ → expectation FAIL dừng pipeline, alert bắn ngay lúc 02:1x thay vì để dữ liệu sai lọt vào báo cáo sáng hôm sau.

Về khía cạnh vận hành/tuân thủ khi đưa các pipeline như vậy lên production ngân hàng, xem Databricks 8 — Ops & Banking.

Ghi nhớ

• Hai lớp bổ sung nhau: Workflows điều phối luồng công việc (job → task → DAG, trigger, retry, alert); DLT lo biến đổi dữ liệu khai báo. Pipeline DLT thường là một task trong Workflows job.
• Workflows: task đa loại (notebook/Python/JAR/SQL/DLT/dbt/run-job), DAG phụ thuộc, trigger cron / file arrival / continuous / API, retry & alert theo task, job cluster dùng chung.
• Workflows vs Airflow: gần như toàn bộ trong Databricks → dùng Workflows tích hợp sẵn; phải điều phối nhiều hệ thống hỗn hợp → Airflow (thường gọi Databricks job).
• DLT khai báo: định nghĩa bảng đầu ra, DLT tự suy DAG, quản incremental/checkpoint/retry. Streaming table cho ingest/làm sạch từng dòng; materialized view cho tổng hợp/join.
• Expectations = data quality inline: EXPECT ... ON VIOLATION warn/drop/fail, phát metrics vào event log.
• Auto Loader nạp file incremental từ cloud storage với schema inference/evolution — nền của lớp bronze.
• Medallion bronze→silver→gold hiện thực tự nhiên bằng DLT; APPLY CHANGES INTO xử lý CDC (dedup, sắp thứ tự theo SEQUENCE BY, delete, SCD Type 1/2).
• Giám sát: DLT event log (trạng thái, DQ metrics, lineage) + lịch sử run của Workflows (luồng, thời lượng, alert).