CI/CD cho Data EngineeringCI/CD for Data Engineering

Thuộc bộ kiến thức Data Engineer Roadmap.

Tổng quan

CI/CD cho một application thông thường trả lời một câu hỏi khá gọn: “code mới có chạy đúng không, và nếu không thì có rollback được không?” Một deploy application tệ thường chỉ có nghĩa là website bị down hoặc một nút bấm bị hỏng — khó chịu, dễ thấy, và gần như luôn có thể khắc phục bằng cách deploy lại phiên bản trước. CI/CD cho một data pipeline phải trả lời một câu hỏi khó hơn, vì artifact được deploy không chỉ phục vụ request — nó ghi dữ liệu. Một deploy dbt model tệ không chỉ đơn giản là chạy fail; nếu nó chạy thành công với logic sai, nó âm thầm ghi đè lên một bảng production với những con số sai, và bảng đó có thể được đọc bởi bốn mươi dashboard downstream, ba machine learning feature, và một báo cáo tài chính gửi lên ban lãnh đạo. Rollback code thì dễ. Rollback lại dữ liệu đã được ghi — cùng mọi bảng derived tính toán từ nó kể từ đó — thì thường không dễ chút nào.

Đây chính là chủ đề trọng tâm của bài viết này: cơ chế của CI/CD (build, test, deploy, verify) giống nhau bất kể bạn đang ship cái gì, và ../devops/en/11-ci-cd.md đã bao quát các cơ chế đó rất kỹ — các stage của pipeline, test pyramid, artifact promotion, so sánh công cụ. Điều thực sự khác biệt ở data engineering là cái gì cần được test trước khi merge (tính đúng đắn của một SQL transformation trên hình dạng dữ liệu thực, chứ không chỉ cú pháp), bao nhiêu phần của project cần được test lại (dbt/Airflow project lớn khiến việc “test toàn bộ trên mọi PR” trở nên quá chậm, đó là lý do có Slim CI), và “blast radius” nghĩa là gì (một deploy tệ làm hỏng một bảng warehouse dùng chung, có trạng thái, thay vì làm crash một service stateless mà chỉ cần redeploy là khắc phục ngay). Bài viết này đi qua CI cho dbt project (bao gồm Slim CI), CI cho Airflow DAG, các chiến lược deployment giới hạn blast radius cho thay đổi dữ liệu, và các quy ước version control đặc thù cho data project — dựa trên nền tảng của ./09-workflow-orchestration.md về những gì đang được orchestrate và ./16-testing-for-data-pipelines.md về các kỹ thuật test mà CI thực sự chạy.

Vì sao “cứ deploy lại là xong” không cứu bạn trong data engineering

Bản năng khiến CI/CD của application trở nên khoan dung — “nếu deploy tệ, rollback container là ổn trong vòng một phút” — sụp đổ với data pipeline vì ba lý do mang tính cấu trúc:

1. State sống lâu hơn deploy. Một application server không giữ state giữa các lần deploy; một bảng warehouse thì có. Nếu một transformation tệ đã chạy và ghi đè dim_customers bằng logic dedupe thực ra là sai, việc revert code dbt về commit trước không hề xóa bỏ dữ liệu tệ đang nằm trong bảng. Bạn cần chạy lại model đã sửa — và đến lúc đó, mọi model downstream tiêu thụ bảng tệ cũng đã tính toán dựa trên nó.
2. Consumer khuếch tán blast radius về phía trước. Một API endpoint bị hỏng fail từng request một, và lỗi dừng lại ngay khi bạn rollback. Một bảng warehouse bị hỏng được đọc bởi mọi dashboard, mọi export theo lịch, mọi ML feature pipeline query nó trong khoảng thời gian giữa lúc chạy tệ và lúc sửa — mỗi consumer đó lại phục vụ những con số sai cho chính khán giả downstream của nó, một số trong đó có thể hành động dựa trên đó (một báo cáo tài chính đã gửi email, một chiến dịch marketing kích hoạt dựa trên một segment sai) trước khi ai đó nhận ra.
3. “Chạy thành công” không có nghĩa là “đúng”. Một SQL transformation với logic join sai một cách tinh vi vẫn chạy tốt, trả về exit code 0, và tạo ra một bảng trông hoàn toàn hợp lý — cho đến khi ai đó ở downstream nhận ra doanh thu bị đếm gấp đôi ba tuần sau đó. Không như một application crash, thường không hề có lỗi nào để alert; failure mode ở đây là dữ liệu sai một cách âm thầm, đó là lý do vì sao test tính đúng đắn của dữ liệu trước khi merge (chứ không chỉ test code compile được) là mối quan tâm trung tâm của CI ở đây.

Không điều nào trong số này có nghĩa là data pipeline cần một toolchain CI/CD hoàn toàn khác — cùng một GitHub Actions, GitLab CI, hay Jenkins build application code cũng build dbt project và Airflow DAG. Điều thay đổi là bạn test cái gì, test lại bao nhiêu phần của project cho mỗi thay đổi, và bạn stage việc rollout chính dữ liệu đó cẩn thận đến mức nào, chứ không chỉ rollout code.

Kiến thức nền tảng

Hình dạng chung của CI/CD, nhắc lại

Như đã nêu trong ../devops/en/11-ci-cd.md, một pipeline chạy build → test → scan → package → deploy → verify, promote cùng một artifact đã test qua dev → staging → production thay vì build lại cho từng environment. Hình dạng đó áp dụng cho data project mà không đổi: một dbt project hay một Airflow DAG repo là code được version-control, được lint, test, và promote qua các environment y hệt một application. Mọi thứ trong bài viết này giả định nền tảng đó và chồng thêm các mối quan tâm đặc thù cho data lên trên — nó không thay thế nền tảng đó. Nếu bạn chưa đọc phần nền tảng về CI vs. delivery vs. deployment, các stage của pipeline, và các chiến lược deployment (blue-green, canary, rolling), hãy đọc bài đó trước.

CI cho dbt project: test SQL trước khi nó chạm vào bảng dùng chung

Về bản chất, một dbt project là một directed graph gồm các câu lệnh SQL SELECT (model) cộng với các assertion về dữ liệu mà những model đó tạo ra (test). Câu hỏi CI tự nhiên là: trước khi merge một thay đổi vào model, SQL mới có thực sự build đúng và pass test không? Câu trả lời đơn giản nhất — chạy dbt build và dbt test cho toàn bộ project với một schema staging, trên mỗi pull request — hoạt động, và với một project nhỏ thì đó chính xác là điều nên làm. dbt build compile và chạy từng model (tôn trọng dependency graph) rồi thực thi các test gắn với nó, nên một câu lệnh duy nhất cho bạn biết cả “có chạy được không” lẫn “output có đúng không”.

Vấn đề là cách này không scale được. Một dbt project với vài trăm model có thể mất hai mươi đến ba mươi phút để rebuild toàn bộ, và nếu một PR chỉ đụng vào một mart ở downstream, việc rebuild toàn bộ số bảng staging tương đương cả warehouse trên mỗi commit là lãng phí thời gian và chi phí. Đây chính xác là trade-off giữa tốc độ feedback và độ tin cậy đã nêu trong ../devops/en/11-ci-cd.md, áp dụng vào SQL: bạn muốn có độ tin cậy rằng thay đổi là đúng, nhưng không muốn trả giá cho việc test code đã được chứng minh là không hề thay đổi.

Slim CI: chỉ test những gì đã thay đổi

Câu trả lời của dbt cho vấn đề này là Slim CI (về mặt kỹ thuật, là selector state:modified kết hợp với deferral). Ý tưởng khá đơn giản khi đã hiểu: dbt có thể so sánh manifest (manifest.json) của project đã compile trong PR hiện tại với manifest của lần chạy production thành công gần nhất, và tính chính xác model nào đã được thêm, thay đổi, hoặc có upstream dependency bị thay đổi. Sau đó bạn yêu cầu dbt chỉ build tập modified đó — cộng với mọi thứ downstream của nó, vì một thay đổi trong một model có thể ảnh hưởng đến bất cứ thứ gì select từ nó:

dbt build --select "state:modified+" --state ./prod-manifest

Dấu + sau state:modified có nghĩa “và mọi thứ downstream của các model đã thay đổi”, điều này rất quan trọng — nếu bạn thay đổi stg_orders, bạn cần rebuild và test lại fct_orders và mart_revenue dù bạn không đụng vào SQL của chúng, vì tính đúng đắn của chúng phụ thuộc vào output của stg_orders. --state trỏ đến một thư mục chứa manifest production để diff; trên thực tế manifest này thường được lấy từ artifact của lần chạy production thành công gần nhất (nhiều team lưu nó trong cloud storage hoặc như một CI artifact sau mỗi lần deploy production).

Slim CI còn dùng thêm deferral (--defer) để các model upstream không thay đổi hoàn toàn không cần được rebuild trong schema CI — dbt resolve các reference đến chúng bằng cách trỏ thẳng vào các bảng production đã được build sẵn, thay vì rebuild một bản copy trong schema CI tạm thời. Điều này có nghĩa một PR chỉ đụng vào một mart có thể validate mart đó trong vài giây thay vì rebuild hàng trăm model upstream mà nó không cần verify lại.

Một workflow GitHub Actions thực tế cho dbt Slim CI

Workflow dưới đây chạy trên mỗi pull request vào một dbt project, cài dependency, lấy manifest production gần nhất làm state để so sánh, và build/test chỉ tập con đã thay đổi trong một schema cô lập, riêng cho từng PR:

# .github/workflows/dbt-ci.yml
name: dbt Slim CI

on:
  pull_request:
    branches: [main]
    paths:
      - "models/**"
      - "macros/**"
      - "tests/**"
      - "dbt_project.yml"

concurrency:
  group: dbt-ci-${{ github.event.pull_request.number }}
  cancel-in-progress: true

env:
  DBT_PROFILES_DIR: ./
  # Isolate each PR's builds into its own schema so concurrent PRs don't collide
  DBT_SCHEMA: ci_pr_${{ github.event.pull_request.number }}

jobs:
  slim-ci:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout PR branch
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dbt
        run: pip install dbt-snowflake==1.8.0   # swap adapter for your warehouse

      - name: Fetch last production manifest
        run: |
          mkdir -p ./prod-manifest
          # manifest.json from the last successful production run,
          # published as a CI artifact or stored in cloud storage
          aws s3 cp s3://our-dbt-artifacts/prod/manifest.json ./prod-manifest/manifest.json
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: dbt deps
        run: dbt deps

      - name: dbt build (Slim CI — modified models + downstream only)
        run: |
          dbt build \
            --select "state:modified+" \
            --state ./prod-manifest \
            --defer \
            --target ci
        env:
          DBT_TARGET_SCHEMA: ${{ env.DBT_SCHEMA }}
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_CI_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_CI_PASSWORD }}

      - name: Post dbt run summary as PR comment
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: 'dbt Slim CI finished — see the job logs for the full build/test summary.'
            })

      - name: Drop CI schema
        if: always()
        run: dbt run-operation drop_schema --args "{schema_name: ${{ env.DBT_SCHEMA }}}"

Vài chi tiết đáng chú ý: filter paths giúp tránh chạy toàn bộ job cho các PR chỉ đụng vào documentation; group concurrency hủy các lần chạy cũ khi PR có commit mới, nhờ vậy CI không lãng phí compute để test một diff đã lỗi thời; schema riêng cho từng PR đảm bảo hai PR đang mở không bao giờ đè lên bảng test của nhau; và bước teardown (if: always()) drop schema tạm ngay cả khi build fail, để schema CI không tích tụ và làm tăng chi phí warehouse. Với những project quá lớn đến mức Slim CI vẫn cảm thấy chậm, một số team thêm bước rebuild toàn bộ theo lịch (chạy dbt build không selector vào ban đêm) như một lớp backstop, để phát hiện độ lệch giữa manifest production và thực tế dù CI ở cấp PR chỉ bao giờ test một subgraph.

CI cho Airflow DAG: bắt lỗi trước khi chúng đến được scheduler

Một Airflow DAG là một file Python mà khi được import phải tạo ra một graph task hợp lệ, không cycle, không lỗi cú pháp — và nó được import bởi DagFileProcessor của scheduler trên mỗi chu kỳ parse, ngay trong production, bất kể bạn có định test nó hay không. Điều này tạo ra một failure mode gần như không tồn tại với application thông thường: một file DAG có bug có thể bị hỏng âm thầm trong Airflow UI (hiện banner import error) trong khi mọi DAG khác trong cùng Airflow instance vẫn chạy bình thường, và không ai để ý cho đến khi có người đi tìm một run chưa bao giờ xuất hiện trong graph view. CI cho DAG tồn tại chính xác để bắt các vấn đề ở thời điểm import và thời điểm cấu trúc này trước khi chúng đến được scheduler.

Test DAG validation

Bài test giá trị cao nhất cho một Airflow codebase là một DAG validation test dạng tổng quát — không phải business logic riêng cho project, chỉ đơn giản là kiểm tra mọi file DAG trong repository import sạch sẽ và tạo ra một graph có cấu trúc hợp lệ:

# tests/test_dag_integrity.py
import os
from airflow.models import DagBag

def test_no_import_errors():
    """Every DAG file in dags/ must parse without raising an exception."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    assert len(dag_bag.import_errors) == 0, (
        f"DAG import failures:\n{dag_bag.import_errors}"
    )

def test_dags_have_no_cycles():
    """DagBag itself raises during load if a cycle exists, but assert explicitly
    that every discovered DAG is a well-formed, non-empty graph."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    assert len(dag_bag.dags) > 0, "No DAGs were discovered — check dag_folder path"
    for dag_id, dag in dag_bag.dags.items():
        assert len(dag.tasks) > 0, f"DAG '{dag_id}' has no tasks"

def test_dags_have_required_tags_and_owner():
    """Enforce team conventions: every DAG must declare an owner for on-call routing."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    for dag_id, dag in dag_bag.dags.items():
        assert dag.default_args.get("owner"), f"DAG '{dag_id}' has no owner set"
        assert dag.tags, f"DAG '{dag_id}' has no tags for filtering in the UI"

DagBag chính là cơ chế mà scheduler của Airflow dùng nội bộ để parse dags/, nên chạy nó trong CI tái tạo chính xác lỗi mà một deploy tệ sẽ gây ra ở production — một import thiếu, một lỗi gõ trong argument của operator, hay một cycle vô tình giữa hai TaskGroup đều lộ ra ở đây, trong một PR check, thay vì trong Airflow UI lúc 2 giờ sáng. Bài test này rẻ (vài giây, không cần external service nào) và nên có mặt trong mọi Airflow repository bất kể quy mô.

Lint DAG

Ngoài tính đúng đắn khi import, static linting bắt các “mùi” cấu trúc trước khi chúng trở thành sự cố production: một task không có upstream hay downstream dependency nào mà lẽ ra phải được nối vào graph, một DAG không có schedule và không chọn catchup một cách tường minh (âm thầm mặc định thành thứ tác giả không hề định), hay một ngày hardcode được dùng thay vì biến template {{ ds }} của Airflow (điều này phá vỡ backfill). Các công cụ như ruff/flake8 bắt các vấn đề Python tổng quát (import không dùng, tên chưa định nghĩa); một lint pass tùy chỉnh hoặc một thư viện như black/isort để format giữ cho file DAG nhất quán trong cả team, điều này quan trọng hơn nghe có vẻ vì file DAG được đọc nhiều hơn hẳn so với module application thông thường — mọi buổi review sự cố đều bắt đầu bằng việc đọc một file DAG.

Test logic của task tách biệt khỏi cấu trúc DAG

Test DAG validation chứng minh graph có cấu trúc hợp lệ; nó không nói gì về việc logic của từng task riêng lẻ có đúng hay không. Những thứ đó cần được test độc lập, giống hệt cách bạn unit test một hàm bất kể ai gọi nó. Nếu một PythonOperator gọi một hàm transform_customer_batch(records), hàm đó nên đủ nhỏ, và đủ tách biệt khỏi runtime context của Airflow, để có thể import và unit test trực tiếp:

# tests/test_transform_logic.py
from dags.customer_pipeline.transforms import transform_customer_batch

def test_transform_customer_batch_deduplicates_by_email():
    raw = [
        {"email": "a@example.com", "name": "A"},
        {"email": "a@example.com", "name": "A (dup)"},
    ]
    result = transform_customer_batch(raw)
    assert len(result) == 1

Sự tách biệt này — cấu trúc DAG được test qua DagBag, business logic được test như những hàm Python thuần túy — là điều giữ cho test suite của Airflow nhanh. Một anti-pattern phổ biến là viết test khởi chạy toàn bộ một DagRun qua scheduler để verify business logic; những test đó chậm, không ổn định, và phần lớn là dư thừa so với những gì một unit test thuần túy của hàm bên dưới đã chứng minh. Hãy dành test ở cấp DAG-run cho hành vi thực sự thuộc về cấp DAG (thứ tự task, cấu hình retry, phân nhánh trigger_rule) và giữ việc test transformation logic ở cấp unit. Về phân loại rộng hơn các loại test đối với chính output của pipeline (schema test, freshness test, kiểm tra bất thường row-count), xem ./16-testing-for-data-pipelines.md.

Khái niệm chính

Chiến lược deployment thích ứng cho data: giảm blast radius của transformation sai

../devops/en/11-ci-cd.md bao quát blue-green và canary deployment cho traffic của application — định tuyến một phần request đến phiên bản mới trước khi commit hoàn toàn. Data pipeline không thể “định tuyến traffic” theo nghĩa tương tự (một bảng không phục vụ live request như một web server), nhưng cùng ý tưởng nền tảng — validate phiên bản mới ở một nơi an toàn, rồi chuyển consumer sang một cách atomic — có những phiên bản tương ứng trực tiếp trong hạ tầng data, và chúng quan trọng chính xác vì việc ghi đè tại chỗ (in-place) là kiểu deploy rủi ro nhất có thể có đối với một bảng dùng chung.

Ghi vào bảng mới rồi swap (phiên bản data của blue-green). Thay vì để một model chạy trực tiếp CREATE OR REPLACE TABLE fct_orders AS SELECT ... lên bảng mà mọi consumer đang đọc, pipeline build phiên bản mới dưới một tên có version hoặc staged — fct_orders__build_20260719 hoặc hậu tố _stg — chạy toàn bộ bộ validation với bảng đó, và chỉ sau khi validation pass mới hoặc rename bảng vào đúng vị trí, hoặc trỏ lại một view/alias ổn định (fct_orders là một view trên bản build đã validate mới nhất) sang phiên bản mới. Nếu validation fail, không có gì ở downstream từng thấy dữ liệu tệ, vì không có gì downstream từng trỏ vào nó. Các warehouse hỗ trợ zero-copy cloning (Snowflake) hay table versioning/time travel (BigQuery, Delta Lake, Iceberg) khiến pattern này rẻ: clone hoặc branch bảng, build trên bản clone, validate, rồi swap alias — phiên bản “cũ” vẫn sẵn sàng như một rollback target tức thì nếu có gì đó bị phát hiện là sai sau khi swap.

Canary theo partition hoặc theo phần trăm. Với các bảng rất lớn hoặc streaming pipeline, một số team rollout transformation mới cho một partition, một customer segment, hoặc một region duy nhất trước, so sánh output của nó với logic cũ chạy song song (một kiểu so sánh “shadow” hay “dual-write”), và chỉ mở rộng rollout khi diff trông hợp lý. Cách này chậm hơn để thiết lập so với cutover thẳng, nhưng cho bạn một sự so sánh trực tiếp với dữ liệu có hình dạng production thay vì hoàn toàn dựa vào fixture test trước khi merge.

Vì sao điều này quan trọng hơn với data so với application: một canary application hoạt động sai chỉ ảnh hưởng đến những user được định tuyến đến nó, trong khoảng thời gian canary còn nhận traffic — blast radius bị giới hạn bởi khối lượng request và thời gian. Một lần ghi đè bảng tại chỗ sai có blast radius không giới hạn theo một chiều khác: mọi job downstream, dashboard, và export đọc bảng đó ở bất kỳ thời điểm nào trước khi ai đó nhận ra và sửa đều thừa hưởng dữ liệu tệ, và mỗi thứ đó, đến lượt mình, có thể đã tạo ra artifact downstream riêng của nó (một snapshot dashboard được cache, một báo cáo đã gửi email, một model được retrain trên feature tệ) mà không tự động được sửa khi bảng nguồn được sửa. Việc stage việc ghi và chỉ swap sau khi validate biến một lỗi không giới hạn, âm thầm thành một lỗi có giới hạn, được bắt trước khi ai đó nhìn thấy — đó chính là toàn bộ mục đích.

Quy ước version control đặc thù cho data project

../backend/en/04-version-control-and-collaboration.md bao quát các nền tảng Git, branching, và PR review áp dụng không đổi cho data repository — dbt project và Airflow DAG repo là các Git repo thông thường, được review qua pull request thông thường. Tuy nhiên có vài quy ước đặc thù cho data project:

• Cấu trúc dbt project như một quy ước, không chỉ là sở thích. Cấu trúc phân lớp được dbt khuyến nghị — staging/ (một model cho mỗi bảng nguồn, chỉ rename/cast nhẹ), intermediate/ (các khối business logic tái sử dụng được), marts/ (bảng cuối cùng, hướng đến nghiệp vụ, nhóm theo domain) — tồn tại để một reviewer có thể đoán được một thay đổi thuộc về đâu và có blast radius gì chỉ từ đường dẫn thư mục của nó. Một thay đổi trong staging/ được kỳ vọng lan rộng (vì mọi thứ đều build trên staging); một thay đổi trong marts/finance/ được kỳ vọng chỉ giới hạn trong một domain. Đi chệch khỏi quy ước này khiến việc review PR khó hơn đáng kể vì reviewer mất đi tín hiệu đó.
• Versioning kiểu migration cho thay đổi schema. Không như application code, nơi một thay đổi schema thường được ẩn sau một công cụ migration của ORM, một thay đổi schema warehouse (thêm cột, đổi kiểu dữ liệu, đổi tên một field được năm model khác tiêu thụ) cần được áp dụng cùng kỷ luật “migration” một cách tường minh: một thay đổi model dbt làm thay đổi hình dạng của một bảng nên được coi là một breaking change đối với một public interface, được thông báo cho consumer, và lý tưởng nhất là rollout theo kiểu cộng thêm (thêm cột mới, migrate consumer, xóa cột cũ ở một PR sau) thay vì một rename atomic duy nhất phá vỡ mọi query downstream ngay khi merge. Một số team theo dõi việc này bằng tag deprecated tường minh trong config meta của dbt và một khoảng deprecation window đã được document trước khi xóa.
• Contracts và việc giới hạn phạm vi bằng dbt build --select như một sự thay thế cho typed interface chặt chẽ. Application code có được đảm bảo tại compile-time từ một type system; SQL model phần lớn thì không, trừ khi bạn dùng config contracts của dbt (config này ép một schema cột/kiểu dữ liệu đã khai báo lên một model tại build time và làm fail build nếu output thực tế của model lệch khỏi đó). Dùng contracts trên các mart được tiêu thụ rộng rãi là điều gần nhất với việc một team application version hóa một public API.

Bức tranh công cụ

Các công cụ CI đa dụng — GitHub Actions, GitLab CI, CircleCI, Jenkins — chính xác là những công cụ chạy pipeline CI cho dbt và Airflow trên thực tế; không có gì trong data engineering đòi hỏi một engine CI khác biệt, và bảng so sánh công cụ trong ../devops/en/11-ci-cd.md áp dụng trực tiếp. Điều đáng biết là CI tích hợp sẵn của dbt Cloud, được thiết kế chuyên biệt cho các team lấy dbt làm trung tâm không muốn tự tay dựng Slim CI trong một pipeline YAML tổng quát: dbt Cloud hỗ trợ sẵn “Slim CI jobs” được trigger trên mỗi PR, tự động diff so với state của lần chạy production thành công gần nhất, dựng lên một schema tạm cho từng PR, và teardown sau đó — về cơ bản chính là toàn bộ workflow GitHub Actions ở trên, nhưng được cấu hình qua giao diện thay vì viết tay YAML. Các team đã đầu tư vào dbt Cloud để schedule production run thường dùng luôn tính năng CI của nó thay vì tự xây tương đương; các team chạy dbt-core mã nguồn mở cùng Airflow hoặc một orchestrator khác thường tự xây tương đương bằng GitHub Actions/GitLab CI, vì không có một control plane dbt Cloud được host riêng để hook vào.

Best Practices

Duy trì các schema dev, staging, và production thực sự tách biệt, và promote giữa chúng một cách có chủ đích. Mỗi developer nên có thể chạy dbt build trên schema dev riêng của họ mà không có rủi ro chạm vào bảng mà ai khác phụ thuộc; CI nên build vào một schema tạm, giới hạn trong phạm vi PR; và production chỉ nên được ghi bởi job production theo lịch chạy từ branch main. Làm mờ ranh giới này — developer test “chỉ lần này thôi” trực tiếp lên schema production, hay CI chia sẻ một schema giữa các PR đang chạy song song — là cách những thay đổi chưa được test rò rỉ vào các bảng mà mọi người phụ thuộc.

Yêu cầu test pass như một merge gate, không phải một kiểm tra mang tính lịch sự. Một CI job chạy dbt test nhưng không chặn merge khi nó fail chỉ mang tính hình thức. Các branch protection rule yêu cầu status check của CI phải pass trước khi một PR được merge vào main biến “chúng ta có test” thành “SQL chưa được test không thể đến được production”, đó mới là sự đảm bảo thực sự bạn muốn có.

Xây dựng khả năng rollback tự động ngay trong bước deploy, không chỉ trong kế hoạch khôi phục. Vì state của bảng tồn tại lâu hơn một lần rollback code, “rollback” đối với một data pipeline có nghĩa nhiều hơn git revert — nó có nghĩa là khả năng trỏ một view/alias trở lại phiên bản bảng đã biết là tốt gần nhất (đó là lý do có pattern ghi-vào-bảng-mới-rồi-swap ở trên), hoặc chạy lại một historical DAG run cụ thể để tái tạo một bảng đã sửa từ nguồn. Hãy thiết kế cơ chế deploy sao cho đây là một lệnh hoặc một nút bấm duy nhất, không phải một quá trình khôi phục ứng biến dưới áp lực lúc 2 giờ sáng.

Không có “cowboy deploy”: không bao giờ sửa tay một file DAG, config của một job dbt đã được lên lịch, hay schema của một bảng production ngoài con đường CI/CD. Luôn có cám dỗ SSH vào một máy scheduler hoặc bấm “edit” trong console của warehouse để hotfix điều gì đó dưới áp lực — và đó chính xác là cách sự lệch pha giữa cái nằm trong Git và cái thực sự đang chạy ở production tích tụ, âm thầm làm mất hiệu lực của cam kết “main branch == cái đang được deploy”. Mọi thay đổi, kể cả fix khẩn cấp, nên đi qua cùng con đường PR → CI → deploy, dù con đường đó có thể cần một “làn hotfix” được đẩy nhanh với review nhẹ hơn (nhưng không bị bỏ qua) cho các sự cố thực sự.

Giữ manifest production dùng cho Slim CI luôn tươi mới. Tính hữu ích của Slim CI hoàn toàn phụ thuộc vào việc so sánh với một bức tranh chính xác về những gì thực sự đang chạy ở production; nếu manifest.json được lưu để dùng làm --state bị cũ (vì một bước deploy quên publish nó, hoặc ai đó merge mà không đi qua CI), Slim CI bắt đầu âm thầm bỏ qua những model thực sự đã thay đổi, điều này phá hỏng toàn bộ mục đích của nó. Publish manifest như một artifact của mỗi lần deploy production thành công — tự động, như một phần của cùng job deploy — là điều giữ cho cơ chế này đáng tin cậy.

Tài liệu tham khảo

• dbt Docs — About continuous integration jobs
• dbt Docs — State comparison and Slim CI selectors (state:modified)
• dbt Docs — Model contracts
• Apache Airflow — Testing DAGs
• Apache Airflow — Best Practices (DAG writing, testing, CI)
• GitHub Actions documentation
• ../devops/en/11-ci-cd.md — nền tảng CI/CD tổng quát và các chiến lược deployment
• ../backend/en/04-version-control-and-collaboration.md — nền tảng Git và collaboration

dbt build --select "state:modified+" --state ./prod-manifest

# .github/workflows/dbt-ci.yml
name: dbt Slim CI

on:
  pull_request:
    branches: [main]
    paths:
      - "models/**"
      - "macros/**"
      - "tests/**"
      - "dbt_project.yml"

concurrency:
  group: dbt-ci-${{ github.event.pull_request.number }}
  cancel-in-progress: true

env:
  DBT_PROFILES_DIR: ./
  # Isolate each PR's builds into its own schema so concurrent PRs don't collide
  DBT_SCHEMA: ci_pr_${{ github.event.pull_request.number }}

jobs:
  slim-ci:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout PR branch
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dbt
        run: pip install dbt-snowflake==1.8.0   # swap adapter for your warehouse

      - name: Fetch last production manifest
        run: |
          mkdir -p ./prod-manifest
          # manifest.json from the last successful production run,
          # published as a CI artifact or stored in cloud storage
          aws s3 cp s3://our-dbt-artifacts/prod/manifest.json ./prod-manifest/manifest.json
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: dbt deps
        run: dbt deps

      - name: dbt build (Slim CI — modified models + downstream only)
        run: |
          dbt build \
            --select "state:modified+" \
            --state ./prod-manifest \
            --defer \
            --target ci
        env:
          DBT_TARGET_SCHEMA: ${{ env.DBT_SCHEMA }}
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_CI_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_CI_PASSWORD }}

      - name: Post dbt run summary as PR comment
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: 'dbt Slim CI finished — see the job logs for the full build/test summary.'
            })

      - name: Drop CI schema
        if: always()
        run: dbt run-operation drop_schema --args "{schema_name: ${{ env.DBT_SCHEMA }}}"

# tests/test_dag_integrity.py
import os
from airflow.models import DagBag

def test_no_import_errors():
    """Every DAG file in dags/ must parse without raising an exception."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    assert len(dag_bag.import_errors) == 0, (
        f"DAG import failures:\n{dag_bag.import_errors}"
    )

def test_dags_have_no_cycles():
    """DagBag itself raises during load if a cycle exists, but assert explicitly
    that every discovered DAG is a well-formed, non-empty graph."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    assert len(dag_bag.dags) > 0, "No DAGs were discovered — check dag_folder path"
    for dag_id, dag in dag_bag.dags.items():
        assert len(dag.tasks) > 0, f"DAG '{dag_id}' has no tasks"

def test_dags_have_required_tags_and_owner():
    """Enforce team conventions: every DAG must declare an owner for on-call routing."""
    dag_bag = DagBag(dag_folder="dags/", include_examples=False)
    for dag_id, dag in dag_bag.dags.items():
        assert dag.default_args.get("owner"), f"DAG '{dag_id}' has no owner set"
        assert dag.tags, f"DAG '{dag_id}' has no tags for filtering in the UI"

# tests/test_transform_logic.py
from dags.customer_pipeline.transforms import transform_customer_batch

def test_transform_customer_batch_deduplicates_by_email():
    raw = [
        {"email": "a@example.com", "name": "A"},
        {"email": "a@example.com", "name": "A (dup)"},
    ]
    result = transform_customer_batch(raw)
    assert len(result) == 1