Series Apache Airflow thực hành · Bài 6/6. Một pipeline chạy tốt trên máy hôm nay vẫn có thể hỏng sau một thay đổi nhỏ: task publish bị gán sai trigger rule, một model SQL đổi công thức hoặc scheduler ngừng chạy mà không tạo ra task failed nào. Bài cuối bổ sung kiểm thử, CI, quy trình triển khai, lịch sử publish và giám sát để những vấn đề đó có tín hiệu rõ ràng.

Chúng ta hoàn thiện project bán hàng của năm bài trước, giữ Airflow 3.1.7 và dbt Core 1.9.8. Bạn sẽ có workflow GitHub Actions kiểm tra trên PostgreSQL dùng một lần và một quy trình release có các bước kiểm chứng. CI không được cấp quyền ghi production; bước triển khai trong bài là quy trình có kiểm soát, chưa phải workflow tự động phát hành vào máy chủ của bạn.

Trước khi bắt đầu, khôi phục fixture gốc: 5 đơn, 3 paid, doanh thu paid 350.000 VND; bỏ intentional_failure.sql của bài 4 và cờ fail.flag của bài 5. Giữ các DAG demo để kiểm thử import, nhưng pause chúng khi không thực hành. Chỉ một DAG được giao nhiệm vụ cập nhật dữ liệu bán hàng tại một thời điểm của lab.

1. Vận hành cần nhìn cả code, hạ tầng và dữ liệu

Health xanh không chứng minh báo cáo đủ đơn. Ngược lại, báo cáo cũ vẫn có thể truy vấn được trong khi scheduler đã dừng. Vì vậy cần cả tín hiệu hệ thống và tín hiệu kết quả. Task callback cũng không tự phát hiện một DAG chưa từng được kích hoạt; không có lần chạy thì không có task failed để callback xử lý.

Mỗi release nên gắn revision Git, tag image và phiên bản dependency. Khi có sự cố, thông tin này giúp phân biệt lỗi mới xuất hiện sau thay đổi code với lỗi đã có từ nguồn. Log và artifact là bằng chứng điều tra; chúng không thay bản sao lưu dữ liệu.

2. Kiểm thử DAG và hợp đồng lỗi bằng pytest

Tạo tests/test_retail.py ở thư mục gốc project, khác dbt-retail/tests chứa SQL test. Những kiểm tra dưới đây không gọi API thật và không ghi database. Chúng xác nhận project đọc được DAG, cổng publish vẫn yêu cầu upstream thành công, dữ liệu sai bị từ chối và chính sách 401/429/503 không bị đảo ngược khi refactor.

import os
from pathlib import Path
import sys
import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'dags'))
os.environ.setdefault('AIRFLOW__CORE__LOAD_EXAMPLES', 'False')

from retail_common import validate_order
from retail_errors import ingest_with_policy


@pytest.fixture(scope='module')
def bag():
    from airflow.models.dagbag import DagBag
    return DagBag(str(ROOT / 'dags'), include_examples=False)


def test_dags_import_and_publish_gate(bag):
    assert bag.import_errors == {}, bag.import_errors
    required = {'retail_hello', 'retail_api_to_postgres', 'retail_daily_orders',
                'retail_dbt_pipeline', 'retail_resilience',
                'retail_pool_demo', 'retail_timeout_demo'}
    assert required <= set(bag.dags)
    pipeline = bag.dags['retail_dbt_pipeline']
    assert pipeline.get_task('publish_report').upstream_task_ids == {'build_dbt'}
    assert pipeline.get_task('publish_report').trigger_rule == 'all_success'


def test_retry_and_pool_contract(bag):
    task = bag.dags['retail_resilience'].get_task('load_with_retry')
    assert task.pool == 'shop_api'
    assert task.retries == 2
    assert task.execution_timeout.total_seconds() == 120
    assert bag.dags['retail_pool_demo'].get_task('heavy').pool_slots == 2


@pytest.mark.parametrize('field,value', [
    ('order_id', True), ('amount_vnd', -1),
    ('amount_vnd', '120000'), ('status', 'unknown'),
    ('created_at', '2026-01-01T09:00:00'),
])
def test_invalid_order_is_rejected(field, value):
    row = {'order_id': 1001, 'amount_vnd': 120000, 'status': 'paid',
           'created_at': '2026-01-01T09:00:00+07:00'}
    row[field] = value
    with pytest.raises((ValueError, TypeError)):
        validate_order(row)


@pytest.mark.parametrize('status,fail_fast', [(401, True), (429, False), (503, False)])
def test_http_retry_policy(monkeypatch, status, fail_fast):
    import requests
    import retail_common
    from airflow.exceptions import AirflowFailException
    response = requests.Response()
    response.status_code = status
    error = requests.HTTPError('test response', response=response)

    def fail():
        raise error

    monkeypatch.setattr(retail_common, 'ingest_orders', fail)
    expected = AirflowFailException if fail_fast else requests.HTTPError
    with pytest.raises(expected):
        ingest_with_policy()

Test import chủ động thêm thư mục dags vào Python path để mô phỏng vị trí module dùng chung trong deployment. Đọc một thư mục tùy ý bằng DagBag trong test không có nghĩa Python tự thêm thư mục đó vào sys.path. Không copy thao tác chèn path này vào mọi DAG; project triển khai cần cấu trúc package và đường dẫn nhất quán.

required kiểm tra đủ tên DAG, tránh trường hợp parser không tìm thấy file nào mà test vẫn pass. Hai assertion của publish là hợp đồng bảo vệ báo cáo: nếu ai đổi trigger rule thành all_done để task luôn chạy, CI phải báo lỗi. Các case dữ liệu dùng giá trị sai biết trước; expected không được tạo bằng cách gọi lại chính logic đang kiểm thử.

Tạo requirements-airflow-ci.txt:

apache-airflow==3.1.7
apache-airflow-providers-postgres==6.5.3

Trong một venv kiểm thử riêng trên Linux/WSL, Python 3.12, cài dependency theo constraints đúng nhánh. Không thực hiện lệnh pip này vào Python của một worker đang phục vụ nếu chưa có quy trình cập nhật image:

python3.12 -m venv .venv-airflow-ci
.venv-airflow-ci/bin/python -m pip install \
  -r requirements-airflow-ci.txt \
  -c https://raw.githubusercontent.com/apache/airflow/constraints-3.1.7/constraints-3.12.txt
.venv-airflow-ci/bin/python -m pip install pytest==8.3.5

PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 AIRFLOW_HOME=/tmp/airflow-retail-ci \
  .venv-airflow-ci/bin/python -m pytest tests/test_retail.py -q

Với các file đúng của series, bộ test có 10 case sau khi pytest mở rộng tham số. Nó chưa chứng minh worker kết nối được broker, dbt chạy được trong image hay quyền database thật đầy đủ. Các lớp đó cần smoke test và integration test riêng; đừng đổi tên một test import thành “đã kiểm thử toàn bộ hệ thống”.

3. CI trên GitHub Actions: dùng database biệt lập

Workflow có hai job độc lập. Job Airflow chạy pytest với đúng bộ phiên bản; job dbt tạo PostgreSQL riêng, nạp fixture rồi build/test và đối soát kết quả. Database dùng một lần không dùng credential production. Một password công khai trong workflow chỉ dành cho container CI sẽ bị hủy sau job, không phải mẫu quản lý mật khẩu thật.

Tạo requirements-dbt-ci.txt:

dbt-core==1.9.8
dbt-postgres==1.9.1

Tạo scripts/ci_database.py. Lệnh load cố ý tạo schema mới trong database CI rỗng, không tự xóa schema đang có để che lỗi môi trường. Lệnh check kiểm tra hai ngày và số tiền tính tay, bổ sung cho phép reconciliation SQL:

import json
import os
from pathlib import Path
import sys
import psycopg2
from psycopg2.extras import execute_values

ROOT = Path(__file__).resolve().parents[1]
with psycopg2.connect(
    host=os.environ['DBT_HOST'], port=int(os.environ['DBT_PORT']),
    dbname=os.environ['DBT_DATABASE'], user=os.environ['DBT_USER'],
    password=os.environ['DBT_PASS'], connect_timeout=5,
) as connection:
    with connection.cursor() as cursor:
        if sys.argv[1] == 'load':
            cursor.execute('create schema raw')
            cursor.execute('''
                create table raw.orders (
                    order_id bigint primary key,
                    created_at timestamptz not null,
                    status text not null check (status in ('paid','pending','cancelled')),
                    amount_vnd bigint not null check (amount_vnd >= 0),
                    loaded_at timestamptz not null default current_timestamp
                )
            ''')
            rows = json.loads((ROOT / 'mock-api/orders.json').read_text())
            execute_values(cursor, '''
                insert into raw.orders (order_id, created_at, status, amount_vnd)
                values %s
            ''', [(r['order_id'],r['created_at'],r['status'],r['amount_vnd']) for r in rows])
        elif sys.argv[1] == 'check':
            cursor.execute('''
                select business_date::text, orders, paid_orders, paid_vnd
                from analytics.mart_daily_sales order by business_date
            ''')
            actual = [(d,int(n),int(p),int(v)) for d,n,p,v in cursor.fetchall()]
            expected = [('2026-01-01',3,2,200000),('2026-01-02',2,1,150000)]
            if actual != expected:
                raise SystemExit(f'Fixture mismatch: {actual}')
            print('Fixture totals passed: 5 orders, 350000 VND paid')
        else:
            raise SystemExit('Use load or check')

Tạo .github/workflows/retail-ci.yml. Cấu trúc repository là project ~/airflow-retail, gồm dags, tests, scripts, mock-api và dbt-retail ở cấp gốc:

name: Retail pipeline CI
on:
  pull_request:
  push:
    branches: [main]
permissions:
  contents: read
jobs:
  airflow-tests:
    runs-on: ubuntu-22.04
    timeout-minutes: 20
    env:
      AIRFLOW_HOME: /tmp/airflow-ci
      AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
      PYTEST_DISABLE_PLUGIN_AUTOLOAD: '1'
    steps:
      - uses: actions/checkout@v6
        with:
          persist-credentials: false
      - uses: actions/setup-python@v6
        with:
          python-version: '3.12'
      - name: Install Airflow with release constraints
        run: |
          curl --fail --location \
            https://raw.githubusercontent.com/apache/airflow/constraints-3.1.7/constraints-3.12.txt \
            --output /tmp/constraints.txt
          python -m pip install -r requirements-airflow-ci.txt -c /tmp/constraints.txt
          python -m pip install pytest==8.3.5
          python -m pip check
      - name: Test imports and contracts
        run: python -m pytest tests/test_retail.py -q
  dbt-tests:
    runs-on: ubuntu-22.04
    timeout-minutes: 15
    env:
      DBT_HOST: 127.0.0.1
      DBT_PORT: '55433'
      DBT_USER: shop
      DBT_PASS: ci_only_disposable_password
      DBT_DATABASE: retail
    services:
      postgres:
        image: postgres:16.6
        env:
          POSTGRES_USER: shop
          POSTGRES_PASSWORD: ci_only_disposable_password
          POSTGRES_DB: retail
        ports:
          - 55433:5432
        options: >-
          --health-cmd "pg_isready -U shop -d retail"
          --health-interval 5s
          --health-timeout 5s
          --health-retries 12
    steps:
      - uses: actions/checkout@v6
        with:
          persist-credentials: false
      - uses: actions/setup-python@v6
        with:
          python-version: '3.12'
      - name: Install isolated dbt dependencies
        run: |
          python -m pip install -r requirements-dbt-ci.txt
          python -m pip check
      - name: Load disposable fixture
        run: python scripts/ci_database.py load
      - name: Build and test dbt
        run: >-
          dbt --log-path "$GITHUB_WORKSPACE/ci-artifacts/logs" build
          --project-dir dbt-retail --profiles-dir dbt-retail/profiles
          --target-path "$GITHUB_WORKSPACE/ci-artifacts/target"
      - name: Check known business totals
        run: python scripts/ci_database.py check
      - name: Preserve diagnostics
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: dbt-ci-diagnostics
          retention-days: 7
          if-no-files-found: warn
          path: ci-artifacts/

Job chạy trực tiếp trên runner nên truy cập PostgreSQL service bằng 127.0.0.1 và cổng map 55433. Nếu chuyển job vào container, hostname/network cần thay tương ứng. Hai job dùng Python environment riêng; không cài dbt và Airflow chung rồi bỏ qua thông báo dependency conflict.

Bổ sung .gitignore để không commit credential, backup hoặc artifact sinh ra trong lúc chạy:

.env
backups/
.venv*/
logs/
__pycache__/
.pytest_cache/
ci-artifacts/
dbt-retail/target/
dbt-retail/logs/

Workflow bảo lưu diagnostics kể cả khi bước build lỗi bằng always(). Artifact có thể chứa SQL và metadata, nên kiểm soát quyền đọc và retention. Major tag của action giúp ví dụ dễ đọc; tổ chức có yêu cầu kiểm soát chuỗi cung ứng nên pin commit SHA đã review và cập nhật có quy trình. Không dùng pull_request_target để chạy mã PR không tin cậy với secret có quyền cao.

Đây là CI kiểm tra code và fixture, chưa tự deploy. Để CI trở thành cổng bắt buộc, repository còn cần thiết lập bảo vệ nhánh và required checks. Không mặc định một file YAML tự ngăn mọi thao tác merge. Hai dòng pin dbt cũng chưa khóa toàn bộ dependency gián tiếp; nên lưu môi trường đã kiểm thử và dùng lockfile phù hợp khi cần tái lập chặt chẽ.

4. Quy trình triển khai: đưa đúng revision vào đúng image

1. Review thay đổi và để cả hai job CI đạt trên revision dự kiến phát hành.
2. Gắn tag image có phiên bản riêng; giữ tag revision trước để có đường quay lại.
3. Xác nhận không có run đang ghi dữ liệu trong lúc thay file/image, tạm pause DAG writer của lab.
4. Build image, phân phối cùng một revision DAG/project tới các thành phần cần dùng và cập nhật cấu hình Compose.
5. Kiểm tra import errors, executable dbt, mount, connection và phiên bản image trước smoke test.
6. Trigger một run, đối soát báo cáo và lịch sử publish; chỉ mở lịch tự động sau khi xác nhận đạt.

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

# Edit AIRFLOW_IMAGE_NAME in .env to the reviewed image tag.
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-scheduler airflow dags list-import-errors
docker compose exec airflow-worker /opt/airflow/dbt-venv/bin/dbt --version

Tag r2 trong ví dụ là nhãn của lần phát hành bạn chủ động tạo, không phải một image sẵn có trên registry. Thay image không tự cập nhật nội dung bind mount nếu thư mục DAG/dbt trên host còn revision cũ. Khi phát hành nhiều máy, phải quản lý phân phối code, dependency và artifact đồng bộ, thay vì sửa thủ công một file trên một worker rồi cho rằng cả cluster đã nhận.

Docker Compose quick-start của series dành cho lab. Có thể học đầy đủ cơ chế ở đây nhưng không xem nó là một cấu hình production đã harden. Triển khai thật còn cần quyền truy cập, secrets backend, kết nối bảo mật, remote logs, sao lưu, tài nguyên và phương án sẵn sàng phù hợp yêu cầu doanh nghiệp.

5. Ghi lịch sử publish trong cùng transaction của báo cáo

Nếu chỉ nhìn loaded_at của raw.orders, bạn không biết báo cáo vừa được publish lại hay đã đứng yên lâu. Bài này thêm ops.publications. Mỗi lần publish thành công ghi run ID, invocation ID của dbt, mốc thời gian và các tổng đã công bố. Nếu transaction publish rollback thì dòng lịch sử cũng không được commit.

Tạo dags/retail_ops.py. Module này chỉ chứa hàm, không tự tạo DAG:

def publish_with_audit(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(*), coalesce(sum(orders),0), coalesce(sum(paid_vnd),0)
                    from analytics.mart_daily_sales
                ''')
                days, orders, paid = cursor.fetchone()
                if days == 0:
                    raise ValueError('Refusing to publish an empty lab mart')
                cursor.execute('create schema if not exists reporting')
                cursor.execute('create schema if not exists ops')
                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('''
                    create table if not exists ops.publications (
                        publication_id bigserial primary key,
                        run_id text not null,
                        dbt_invocation text not null,
                        published_at timestamptz not null default current_timestamp,
                        total_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
                ''')
                cursor.execute('''
                    insert into ops.publications
                        (run_id, dbt_invocation, total_orders, paid_vnd)
                    values (%s,%s,%s,%s) returning publication_id
                ''', (build['run_id'], build['invocation_id'], int(orders), paid))
                publication_id = cursor.fetchone()[0]
        return {'publication_id': publication_id, 'orders': int(orders),
                'paid_vnd': str(paid)}
    finally:
        connection.close()

Trong retail_pipeline.py của bài 4, thay toàn bộ định nghĩa task publish_report cũ bằng đoạn sau, vẫn giữ lời gọi nối dependency ở cuối DAG:

@task(retries=0)
    def publish_report(build):
        from retail_ops import publish_with_audit
        return publish_with_audit(build)

Đây là đoạn thay bên trong hàm retail_dbt_pipeline, không phải một file DAG độc lập. Với fixture gốc, một lần publish phải có total_orders=5 và paid_vnd=350000 trong ops.publications. Clear riêng task publish có thể tạo thêm một bản ghi lịch sử cho lần công bố lại; dữ liệu báo cáo không bị nhân đôi vì vẫn thay nội dung một bảng trong transaction.

Mốc published_at là thời điểm transaction ghi nhận bản công bố, không phải ngày nghiệp vụ của đơn hàng. Lịch sử này ghi các lần publish thành công, chưa ghi mọi lần thử failed. Muốn audit thất bại, cần kết hợp trạng thái DAG/task và log; đừng gọi bảng này là nhật ký đầy đủ của mọi sự kiện vận hành.

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 publication_id, run_id, published_at, total_orders, paid_vnd
    from ops.publications order by publication_id desc limit 5;"

6. Giám sát: sức khỏe hệ thống và tuổi bản báo cáo

6.1. Health endpoint phải đọc nội dung JSON

Airflow 3.1.7 có endpoint /api/v2/monitor/health. HTTP 200 cho biết REST call thành công, không chứng minh mọi thành phần khỏe. Phải đọc trạng thái bên trong payload. Tạo scripts/check_health.py cho ba thành phần cần trong lab:

import json
import sys

payload = json.load(sys.stdin)
required = ('metadatabase', 'scheduler', 'dag_processor')
bad = {key:payload.get(key,{}).get('status') for key in required
       if payload.get(key,{}).get('status') != 'healthy'}
if bad:
    raise SystemExit(f'Airflow components are not healthy: {bad}')
print('Required Airflow components are healthy')

set -o pipefail
curl --fail --silent --show-error \
  http://localhost:8080/api/v2/monitor/health \
  | python scripts/check_health.py

Lệnh chạy từ host có Python phù hợp và quyền truy cập endpoint. Nếu deployment yêu cầu xác thực, dùng cơ chế được cấp quyền của hệ thống, không tắt auth để tiện monitor. Triggerer không được đưa vào danh sách bắt buộc của script vì có deployment không sử dụng deferrable task; nếu pipeline phụ thuộc triggerer, cần giám sát nó tương ứng.

Health còn phụ thuộc đường truy cập API. Nếu API server hỏng, bạn có thể không đọc được trạng thái scheduler dù scheduler vẫn chạy. Kết hợp CLI check hoặc health server riêng của scheduler và giám sát worker/broker. Một request lỗi mạng cần tạo tín hiệu riêng thay vì bị coi là payload healthy trống.

6.2. Kiểm tra tuổi bản publish thay vì chỉ tìm task failed

Tạo scripts/check_publish.py. Script dùng DSN lab trong AIRFLOW_CONN_SHOP_POSTGRES nhưng kết nối bằng psycopg2 trực tiếp, không cần giả lập task context của SDK. Nó không in DSN:

import os
import psycopg2

maximum = int(os.environ.get('MAX_PUBLISH_AGE_MINUTES', '120'))
if maximum <= 0:
    raise SystemExit('MAX_PUBLISH_AGE_MINUTES must be positive')
with psycopg2.connect(os.environ['AIRFLOW_CONN_SHOP_POSTGRES'], connect_timeout=5) as connection:
    with connection.cursor() as cursor:
        cursor.execute('''
            select publication_id, run_id, published_at,
                   extract(epoch from (current_timestamp - published_at))/60 as age_minutes,
                   total_orders, paid_vnd
            from ops.publications order by publication_id desc limit 1
        ''')
        row = cursor.fetchone()
        if row is None:
            raise SystemExit('No publication has been recorded')
        publication_id, run_id, published_at, age, count, paid = row
        print({'publication_id':publication_id, 'run_id':run_id,
               'published_at':str(published_at), 'age_minutes':float(age),
               'orders':count, 'paid_vnd':str(paid)})
        if age < 0 or age > maximum:
            raise SystemExit('Publication is stale or clock is inconsistent')

Để thử trong worker, bổ sung mount scripts vào service airflow-worker trong compose.dbt.yaml, cùng cấp với mount dbt-retail đang có. Tạo thư mục scripts trước khi up:

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

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

docker compose exec -e MAX_PUBLISH_AGE_MINUTES=120 airflow-worker \
  python /opt/airflow/scripts/check_publish.py

Đây mới là phép kiểm tra chạy tay. Trong vận hành thật, monitor phải chạy từ nơi độc lập với scheduler/worker cần giám sát và gửi cảnh báo qua kênh tổ chức đã thiết lập. Nếu worker chết, một script chỉ được gọi bên trong chính worker không thể tự phát hiện toàn bộ sự cố đó.

Tuổi publish chỉ cho biết báo cáo vừa được công bố, chưa chứng minh nguồn đã có đủ dữ liệu nghiệp vụ mới. Một pipeline có thể tính lại batch cũ và vẫn tạo publication mới. Trong hệ thống thật, kết hợp watermark hoặc batch ID upstream, kiểm tra độ đầy đủ và đối soát nghiệp vụ; không dùng timestamp publish thay mọi loại freshness.

Ngưỡng 120 phút chỉ hợp lý khi pipeline được kỳ vọng refresh khoảng mỗi giờ. Với DAG đang chạy tay, báo cáo quá 120 phút là bình thường nếu bạn chưa đặt cam kết lịch. Nếu muốn lab tự refresh hàng giờ, thay schedule=None của retail_dbt_pipeline bằng timetable kích hoạt dưới đây, giữ catchup=False và pause các writer cũ:

from airflow.timetables.trigger import CronTriggerTimetable

schedule=CronTriggerTimetable(
    '15 * * * *', timezone='Asia/Ho_Chi_Minh'
)

Đoạn này thay phần schedule trong decorator, không tạo DAG thứ hai. Pipeline này full scan và build toàn bộ mart nhỏ, nên lịch chỉ là mốc kích hoạt; nó không dùng data interval làm bộ lọc ngày. Nếu chuyển sang xử lý từng ngày, dùng thiết kế interval ở bài 3. Không lấy hai ngữ nghĩa lịch khác nhau rồi trộn vào cùng một phép chọn dữ liệu.

7. Giao diện thực hành: đọc Home và Health đúng ý nghĩa

1. Mở Home để xem các chỉ báo health và số run/task theo phạm vi thời gian đang chọn.
2. Từ danh sách lỗi, mở đúng DAG và run để điều tra task cụ thể; đừng chỉ nhìn tỷ lệ success tổng.
3. Đối chiếu run ID đó với ops.publications nếu cần biết bản báo cáo nào đã được công bố.
4. Dùng monitor bên ngoài để kiểm tra tuổi publish; Home xanh không thay phép kiểm tra dữ liệu còn đúng hạn.

Một ngày không có run mới có thể không tạo ra một chấm đỏ nào trên dashboard. Vì vậy điều kiện “không có bản publish trong khoảng cho phép” phải được theo dõi rõ, ngoài các thông báo exception của task.

8. Diễn tập sự cố và khôi phục có kiểm soát

Rollback Git chỉ khôi phục code. Nó không tự mang lại các dòng đã bị xóa, trạng thái snapshot cũ hay một bảng đã full-refresh sai. Với bảng báo cáo nhỏ của series có thể rebuild từ raw nếu raw đầy đủ và quy tắc đúng. Nếu raw cũng mất hoặc đã bị sửa, cần backup hoặc nguồn có thể tái tạo, không thể chỉ chạy lại DAG để suy ra lịch sử chưa từng được lưu.

Tạo backup database lab trước khi thử một thay đổi phá cấu trúc. Lệnh dưới chỉ đọc database vào file backup, không xóa dữ liệu:

mkdir -p backups
(
  set -e
  set -o noclobber
  backup_path="backups/retail-$(date -u +%Y%m%dT%H%M%SZ).dump"
  docker compose -f docker-compose.yaml -f compose.override.yaml \
    -f compose.dbt.yaml exec -T shop-db pg_dump -U shop -d retail -Fc \
    > "$backup_path"
  test -s "$backup_path"
  printf 'Backup file: %s\n' "$backup_path"
)

File có dung lượng không chứng minh restore được. Nghiệm thu backup bằng cách phục hồi sang một database hoặc môi trường thử riêng, kiểm tra schema, số dòng và tổng tiền. Không dán lệnh restore đè lên database đang phục vụ chỉ vì tên file backup có vẻ đúng. Backup cũng có dữ liệu nhạy cảm nên cần bảo vệ quyền đọc, retention và bản lưu ngoài host phù hợp.

Runbook tối thiểu ghi người phụ trách, tác động, run ID, revision/image, lỗi đầu tiên, bản báo cáo đang phục vụ, phạm vi chạy lại và cách đối soát. Sau khi phục hồi cần kiểm tra cache BI nếu có; người xem có thể còn nhìn bản cũ dù warehouse đã cập nhật. Không xóa lịch sử lỗi để “dọn sạch” trước khi hoàn tất điều tra.

9. Nghiệm thu project cuối series

1. Từ repository đầy đủ, chạy 10 test Python và kiểm tra không có import error hoặc DAG thiếu.
2. Cho CI tạo database mới, nạp fixture, chạy dbt build/test và xác nhận 5 đơn, 350.000 VND.
3. Triển khai một revision có tag image rõ ràng và thực hiện smoke test trên lab.
4. Publish thành công, tìm run ID và invocation ID tương ứng trong ops.publications.
5. Tạo lỗi dbt có chủ đích; xác nhận reporting và publication mới không được commit khi gate chưa đạt.
6. Kiểm tra health payload có thành phần unhealthy; script phải trả mã khác 0 dù HTTP call thành công.
7. Thử ngưỡng tuổi publish trong môi trường thử, xác nhận cảnh báo stale hoạt động và hiểu vì sao cần monitor độc lập.
8. Tạo backup, diễn tập restore ở môi trường riêng và đối soát dữ liệu trước khi coi phương án khôi phục là sẵn sàng.

Qua sáu bài, bạn đã đi từ DAG đầu tiên tới API có phân trang, PostgreSQL upsert, lịch theo ngày dữ liệu, dbt, cổng chất lượng, retry/timeout/Pools và quy trình vận hành. Phần quan trọng là hiểu mỗi cơ chế bảo vệ điều gì và giới hạn nằm ở đâu. Chỉ thêm công cụ khi nó giải quyết một vấn đề đo được của pipeline.

Ôn lại bài 4: Airflow kết hợp dbt, bài 5: Retry, Timeout và Pools, hoặc bắt đầu từ bài 1: Airflow từ đầu.

Tài liệu tham khảo

• Airflow best practices và testing
• Kiểm tra sức khỏe Airflow
• Airflow UI Overview
• GitHub Actions PostgreSQL service
• Airflow constraints 3.1.7
• PostgreSQL pg_dump
• pytest monkeypatch