Series Apache Airflow thực hành · Bài 4/6. Airflow đã lấy được đơn hàng từ API vào PostgreSQL, nhưng báo cáo vẫn cần một lớp SQL rõ ràng và các phép kiểm tra trước khi cho người dùng xem. Trong bài này, bạn sẽ kết hợp Airflow với dbt để xây một pipeline bán hàng hoàn chỉnh: nạp nguồn, kiểm tra batch, tạo bảng phân tích, chạy test và cập nhật lớp báo cáo khi đạt yêu cầu.

Điểm cần kiểm chứng là dữ liệu sai phải chặn bước publish. Một DAG có bốn task màu xanh chưa đủ nếu phép cộng doanh thu sai; ngược lại, dbt báo lỗi không có nghĩa mọi bảng đã tự quay về trạng thái cũ. Bài sẽ dùng riêng schema analytics cho kết quả chuẩn bị và reporting cho bản báo cáo đã được chấp nhận.

Tiếp tục project ~/airflow-retail ở bài 1–3, giữ Airflow 3.1.7, PostgreSQL 16.6 và thêm dbt Core 1.9.8/dbt-postgres 1.9.1. Dataset gốc có 5 đơn, 3 đơn paid và doanh thu paid 350.000 VND. Nếu đã thử đơn đến muộn 1006 ở bài 3, hoàn nguyên fixture và bản ghi thử trước khi đối soát các số trong bài.

1. Airflow điều phối, dbt quản lý biến đổi SQL

Airflow chịu trách nhiệm quyết định bước nào được chạy, ghi trạng thái và cho người vận hành theo dõi. dbt tổ chức model SQL, quan hệ phụ thuộc giữa model và các phép kiểm tra dữ liệu. Trong một task Airflow, chúng ta gọi dbt bằng executable riêng; dbt kết nối PostgreSQL để thực thi SQL, không chuyển toàn bộ đơn hàng về Airflow để cộng bằng Python.

Để tập trung vào tích hợp, DAG mới chạy tay và full scan năm đơn của lab. DAG theo ngày ở bài 3 vẫn là ví dụ riêng. Pause các DAG cũ trước khi thử DAG mới, tránh nhiều writer cùng sửa raw.orders hoặc báo cáo trong lúc đối soát. Khi mở rộng sang lịch ngày, hãy truyền interval vào ingestion như bài 3 và thiết kế phạm vi build/publish tương ứng, thay vì vừa chạy toàn bảng vừa gọi đó là xử lý một ngày.

2. Cài dbt trong môi trường riêng bên trong image Airflow

dbt và Airflow có hệ dependency riêng. Cài chung tùy tiện vào cùng Python environment có thể làm thay đổi package của Airflow. Lab tạo một virtual environment dành riêng cho dbt trong image; task gọi đường dẫn tuyệt đối nên không cần activate venv bằng shell.

Tạo Dockerfile tại thư mục gốc project:

FROM apache/airflow:3.1.7
USER airflow
RUN python -m venv /opt/airflow/dbt-venv \
    && PIP_USER=false /opt/airflow/dbt-venv/bin/pip install --no-cache-dir \
       dbt-core==1.9.8 dbt-postgres==1.9.1

PIP_USER=false chỉ áp dụng cho lệnh pip trong venv này. Điều đó tránh thiết lập user-install của base image gây lỗi khi cài vào một venv riêng. Các phiên bản được pin để thống nhất bài lab; chúng không thay thế quy trình cập nhật bản vá cho môi trường triển khai thật.

cd ~/airflow-retail
docker build -t techdata/airflow-retail:3.1.7-dbt1.9.8 .

Mở .env, thay dòng AIRFLOW_IMAGE_NAME đang có bằng giá trị dưới đây; giữ AIRFLOW_UID và các dòng khác. Không ghi đè cả file chỉ để đổi một biến:

AIRFLOW_IMAGE_NAME=techdata/airflow-retail:3.1.7-dbt1.9.8

Tạo compose.dbt.yaml. File bổ sung chỉ mount project dbt; các volume dags/logs từ file Compose gốc vẫn được giữ theo cơ chế merge:

services:
  airflow-worker:
    volumes:
      - ./dbt-retail:/opt/airflow/dbt-retail:ro

mkdir -p dbt-retail/models dbt-retail/tests dbt-retail/profiles

docker compose -f docker-compose.yaml -f compose.override.yaml \
  -f compose.dbt.yaml config --quiet

docker compose -f docker-compose.yaml -f compose.override.yaml \
  -f compose.dbt.yaml up -d

docker compose exec airflow-worker /opt/airflow/dbt-venv/bin/dbt --version

Mount project read-only giúp task không vô tình sửa SQL đang triển khai. dbt vẫn cần ghi log và artifact, nên DAG sẽ đưa chúng sang /opt/airflow/logs/dbt, thuộc volume logs đã có. Dùng đủ ba file -f cho thao tác cập nhật stack từ bài này trở đi. Image phải được build xong trước khi up, nếu không Docker có thể tìm một tag local chưa tồn tại.

3. Project dbt nhỏ nhưng có model, source và test

Tạo dbt-retail/dbt_project.yml:

name: retail_airflow
version: '1.0.0'
config-version: 2
profile: retail_airflow
model-paths: ['models']
test-paths: ['tests']
models:
  retail_airflow:
    +materialized: view

Tạo dbt-retail/profiles/profiles.yml. Profile chỉ đọc biến môi trường khi dbt chạy. Credential thật được task lấy từ connection shop_postgres và truyền cho tiến trình dbt; không được in toàn bộ environment ra log:

retail_airflow:
  target: lab
  outputs:
    lab:
      type: postgres
      host: "{{ env_var('DBT_HOST') }}"
      port: "{{ env_var('DBT_PORT', '5432') | int }}"
      user: "{{ env_var('DBT_USER') }}"
      password: "{{ env_var('DBT_PASS') }}"
      dbname: "{{ env_var('DBT_DATABASE') }}"
      schema: analytics
      threads: 2
      connect_timeout: 10

Schema analytics là nơi chuẩn bị kết quả. Khi dùng cấu hình schema mặc định này, hai model bên dưới nằm trong analytics. Project retail_airflow ở bài này khác project dbt của một series khác; không trộn các file source có schema hay cột không tương ứng với raw.orders của lab Airflow.

Tạo dbt-retail/models/sources.yml. unique/not_null kiểm tra khóa và cột bắt buộc; accepted_values giới hạn trạng thái được tính toán:

version: 2
sources:
  - name: shop
    schema: raw
    tables:
      - name: orders
        columns:
          - name: order_id
            tests: [unique, not_null]
          - name: status
            tests:
              - not_null
              - accepted_values:
                  values: ['paid', 'pending', 'cancelled']
          - name: created_at
            tests: [not_null]
          - name: amount_vnd
            tests: [not_null]
models:
  - name: mart_daily_sales
    columns:
      - name: business_date
        tests: [unique, not_null]
      - name: paid_vnd
        tests: [not_null]

Tạo dbt-retail/models/stg_orders.sql:

select
    order_id,
    (created_at at time zone 'Asia/Ho_Chi_Minh')::date as business_date,
    status,
    amount_vnd
from {{ source('shop', 'orders') }}

business_date được tính theo múi giờ Việt Nam, không theo timezone của session SQL. Một đơn gần nửa đêm có thể có ngày UTC khác ngày bán hàng, nên giữ quy tắc này giống bài 3.

Tạo dbt-retail/models/mart_daily_sales.sql:

{{ config(materialized='table') }}
select
    business_date,
    count(*)::bigint as orders,
    count(*) filter (where status = 'paid')::bigint as paid_orders,
    coalesce(sum(amount_vnd) filter (where status = 'paid'), 0)::numeric(20,0) as paid_vnd
from {{ ref('stg_orders') }}
group by business_date

orders là số đơn thuộc mọi trạng thái của ngày, paid_orders chỉ đếm paid và paid_vnd chỉ cộng paid. amount_vnd đã là tiền nguyên VND; kết quả SUM được giữ numeric để tránh giới hạn kiểu integer nhỏ. Một ngày không có đơn sẽ không tự có dòng trong mart này; muốn hiển thị ngày rỗng cần bảng lịch và quy tắc bổ sung.

Tạo dbt-retail/tests/assert_reconciliation.sql. Một singular data test trả về các dòng vi phạm; không có dòng trả về là đạt:

with expected as (
    select
        (created_at at time zone 'Asia/Ho_Chi_Minh')::date as business_date,
        count(*) as orders,
        count(*) filter (where status = 'paid') as paid_orders,
        coalesce(sum(amount_vnd) filter (where status = 'paid'), 0) as paid_vnd
    from {{ source('shop', 'orders') }}
    group by 1
), actual as (
    select * from {{ ref('mart_daily_sales') }}
)
select coalesce(e.business_date, a.business_date) as business_date
from expected e
full outer join actual a using (business_date)
where e.orders is distinct from a.orders
   or e.paid_orders is distinct from a.paid_orders
   or e.paid_vnd is distinct from a.paid_vnd

FULL OUTER JOIN phát hiện cả ngày bị thiếu ở mart lẫn ngày thừa không có ở nguồn. IS DISTINCT FROM so sánh được cả trường hợp null. Test này kiểm tra mart có khớp phép tổng hợp trực tiếp từ raw theo quy tắc đã chọn, nhưng không chứng minh nguồn đã chứa đủ mọi đơn của doanh nghiệp. Fixture biết trước 350.000 VND vẫn là một đối soát độc lập cần thực hiện.

Không dùng loaded_at hiện có làm heartbeat cho từng batch: code upsert ở bài 2 cố ý không cập nhật bản ghi khi nội dung không đổi. Vì vậy một lần nạp thành công không nhất thiết khiến timestamp của tất cả đơn trở nên mới. Bài dùng metadata batch và task dependency để kiểm tra đầu vào của run; freshness và độ đầy đủ của một nguồn bên ngoài cần contract riêng.

4. DAG Airflow gọi dbt và chỉ publish khi kiểm tra đạt

Tạo dags/retail_pipeline.py. Giữ module retail_common.py của bài 2, vì task nạp dùng lại hàm ingest_orders. Module dùng chung không được tự tạo một DAG khi import.

from datetime import timedelta
import pendulum
from airflow.sdk import dag, task, get_current_context
from retail_common import ingest_orders


@dag(
    dag_id='retail_dbt_pipeline', schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz='Asia/Ho_Chi_Minh'),
    catchup=False, max_active_runs=1,
    tags=['techdata', 'retail', 'lesson-4'],
)
def retail_dbt_pipeline():
    @task(retries=1, retry_delay=timedelta(seconds=20),
          execution_timeout=timedelta(minutes=5))
    def load_source():
        return ingest_orders()

    @task(retries=0)
    def check_source(batch):
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        connection = PostgresHook(postgres_conn_id='shop_postgres').get_conn()
        try:
            with connection.cursor() as cursor:
                cursor.execute('select count(*) from raw.orders')
                count = cursor.fetchone()[0]
            if batch['fetched_orders'] == 0 or count != batch['fetched_orders']:
                raise ValueError('Full-scan batch is empty or does not match raw row count')
            return batch
        finally:
            connection.close()

    @task(retries=0, execution_timeout=timedelta(minutes=10))
    def build_dbt(batch):
        import hashlib
        import json
        import os
        from pathlib import Path
        import subprocess
        import uuid
        from airflow.providers.postgres.hooks.postgres import PostgresHook

        context = get_current_context()
        run_id = context['run_id']
        run_hash = hashlib.sha256(run_id.encode()).hexdigest()[:16]
        output = Path('/opt/airflow/logs/dbt') / run_hash / uuid.uuid4().hex[:12]
        output.mkdir(parents=True, exist_ok=False)
        connection = PostgresHook.get_connection('shop_postgres')
        env = os.environ.copy()
        env.update({
            'DBT_HOST': connection.host,
            'DBT_PORT': str(connection.port or 5432),
            'DBT_USER': connection.login,
            'DBT_PASS': connection.password,
            'DBT_DATABASE': connection.schema,
        })
        subprocess.run([
            '/opt/airflow/dbt-venv/bin/dbt', '--no-use-colors',
            '--log-path', str(output / 'logs'), 'build',
            '--project-dir', '/opt/airflow/dbt-retail',
            '--profiles-dir', '/opt/airflow/dbt-retail/profiles',
            '--target-path', str(output / 'target'),
        ], env=env, check=True, timeout=540)
        result = json.loads((output / 'target/run_results.json').read_text())
        statuses = [r['status'] for r in result.get('results', [])]
        if not statuses or any(s in {'error', 'fail', 'skipped'} for s in statuses):
            raise ValueError('dbt results did not pass the publication gate')
        return {'run_id': run_id, 'artifact_dir': str(output),
                'invocation_id': result['metadata']['invocation_id'],
                'fetched_orders': batch['fetched_orders']}

    @task(retries=0)
    def publish_report(build):
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        connection = PostgresHook(postgres_conn_id='shop_postgres').get_conn()
        try:
            with connection:
                with connection.cursor() as cursor:
                    cursor.execute('select count(*) from analytics.mart_daily_sales')
                    if cursor.fetchone()[0] == 0:
                        raise ValueError('Refusing to publish an empty lab mart')
                    cursor.execute('create schema if not exists reporting')
                    cursor.execute('''
                        create table if not exists reporting.daily_sales (
                            business_date date primary key,
                            orders bigint not null,
                            paid_orders bigint not null,
                            paid_vnd numeric(20,0) not null
                        )
                    ''')
                    cursor.execute('delete from reporting.daily_sales')
                    cursor.execute('''
                        insert into reporting.daily_sales
                            (business_date, orders, paid_orders, paid_vnd)
                        select business_date, orders, paid_orders, paid_vnd
                        from analytics.mart_daily_sales
                    ''')
            print({'published_run': build['run_id'],
                   'dbt_invocation': build['invocation_id']})
        finally:
            connection.close()

    publish_report(build_dbt(check_source(load_source())))

retail_dbt_pipeline()

subprocess.run dùng danh sách tham số, check=True và timeout. dbt trả mã lỗi sẽ làm task build_dbt thất bại; không có “|| true” để che mã thoát. Các chuỗi run_id không được nối vào shell command. Đường dẫn artifact dùng hash và mã ngẫu nhiên để tránh ghi đè giữa các lần chạy hoặc retry.

Mỗi lần build đọc đúng run_results.json trong thư mục vừa tạo. Nếu dbt lỗi trước khi tạo artifact, check=True đã dừng task; không lấy file target cũ làm bằng chứng. Build trả XCom ngắn gồm invocation ID, thư mục artifact và run ID, không trả nội dung manifest hay toàn bộ dữ liệu.

publish_report đọc relation PostgreSQL, không dựa vào việc worker khác nhìn thấy file cục bộ của task build. Mặc định trigger rule cần upstream thành công. Chỉ một transaction thay dữ liệu reporting.daily_sales; nếu INSERT lỗi thì DELETE của cùng transaction cũng được rollback. BI nên đọc reporting.daily_sales, không đọc trực tiếp bảng analytics đang được dựng lại.

5. Giao diện thực hành: đọc đường đi lỗi trong Graph View

1. Trong UI, mở DAG retail_dbt_pipeline và chọn đúng run vừa Trigger.
2. Mở Graph để kiểm tra thứ tự bốn task; publish_report phải ở sau build_dbt.
3. Nếu một task lỗi, chọn task đó và mở Logs, đối chiếu Try Number và run ID.
4. Phân biệt task trực tiếp failed với task downstream upstream_failed; điều tra lỗi đầu tiên thay vì bắt đầu từ publish bị chặn.

Với DAG của bài, khi build_dbt failed thì publish_report chưa được phép thực thi. Các số liệu và tên task trong ảnh mẫu không phải số liệu lab; mục đích của ảnh là giúp bạn nhận diện vị trí Graph và bảng Task Instances trên giao diện thật.

6. Chạy pipeline và đối soát bằng SQL

docker compose exec airflow-scheduler airflow dags pause retail_api_to_postgres
docker compose exec airflow-scheduler airflow dags pause retail_daily_orders
docker compose exec airflow-scheduler airflow dags list-import-errors
docker compose exec airflow-scheduler airflow dags unpause retail_dbt_pipeline
docker compose exec airflow-scheduler airflow dags trigger retail_dbt_pipeline

Chờ DAG processor nhận file mới trước khi Trigger. Nếu task báo không tìm thấy executable dbt, kiểm tra image đang chạy của worker và đường dẫn venv; việc dbt đã cài trên host không giúp executable xuất hiện tự động trong container.

docker compose -f docker-compose.yaml -f compose.override.yaml \
  -f compose.dbt.yaml exec -T shop-db psql -U shop -d retail \
  -c "select * from reporting.daily_sales order by business_date;"

Tổng paid phải bằng 350.000 VND. Chạy lại fixture không đổi, bảng báo cáo vẫn hai ngày và cùng tổng tiền. Kiểm tra log build_dbt có invocation ID của đúng lần chạy; đừng chỉ mở thư mục artifact gần giống tên rồi kết luận kiểm tra đã pass.

7. Thử lỗi để chứng minh báo cáo cũ được giữ lại

Sau một lần thành công, tạo tạm dbt-retail/tests/intentional_failure.sql với đúng một dòng vi phạm. Đây là phép thử trong lab:

select 1 as intentionally_failing_row

Tạm đổi đơn 1005 trong orders.json từ pending thành paid, rồi Trigger run mới. Raw có thể được cập nhật và bảng analytics có thể đã được dựng, nhưng singular test cố ý trả một dòng khiến dbt build lỗi. publish_report phải bị chặn; reporting.daily_sales vẫn giữ tổng đã duyệt 350.000 VND.

Xóa đúng file intentional_failure.sql vừa tạo và Trigger lại. Lần này test đạt và reporting phải lên 420.000 VND. Cuối cùng khôi phục 1005 thành pending rồi chạy lại để về fixture gốc. Không xóa cả thư mục tests chỉ để một bài thử hết báo lỗi.

Thử nghiệm này tách rõ ba khái niệm: task load đã commit, model analytics có thể đã thay đổi, còn bảng reporting chưa publish. Trạng thái failed của DAG không tự rollback tất cả task trước. Bản báo cáo cũ cũng cần có thời điểm cập nhật để người xem biết đang đọc phiên bản nào; việc bổ sung lịch sử publish và kiểm tra độ trễ sẽ có ở bài 6.

8. Những giới hạn cần hiểu trước khi mở rộng

Transaction publish bảo vệ một bảng đích của ví dụ, không tạo một giao dịch nguyên khối cho toàn warehouse hay nhiều dashboard. Một report chạy nhiều query ở các thời điểm khác nhau vẫn cần thiết kế tính nhất quán riêng. Bài cũng chưa tự refresh cache BI; dữ liệu warehouse mới không bảo đảm người xem đã nhìn thấy số mới ngay.

9. Bài lab nghiệm thu

1. Worker gọi được dbt trong venv riêng; project mount read-only và artifact ghi vào logs.
2. Chạy đủ bốn task; đối soát hai dòng ngày, năm đơn và 350.000 VND paid.
3. Tạo test cố ý fail, xác nhận build_dbt lỗi và publish_report bị chặn.
4. Đổi trạng thái 1005 trong lúc test đang fail, xác nhận reporting vẫn giữ bản cũ.
5. Bỏ test lỗi, chạy lại và xác nhận publish lên 420.000; sau đó hoàn nguyên fixture về 350.000.
6. Dùng UI tìm đúng run, task trực tiếp failed, log và invocation ID của dbt.

Đọc lại bài 3 về lịch và data interval hoặc bài 2 về API và upsert nếu cần ôn phần dữ liệu nguồn. Bài tiếp theo sẽ đi sâu vào retry, timeout và Pools để một sự cố tạm thời không trở thành hàng loạt task chạy chồng lên nhau.

Tài liệu tham khảo

• dbt build
• dbt data tests
• dbt artifacts
• Airflow best practices
• Giao diện Airflow 3.1.7
• Xây custom Docker image Airflow