Apache Airflow: Xây dựng và vận hành Data Pipeline thực tế · Bài 2/6. Ở bài trước, DAG lấy ba đơn từ một list Python. Bài này thay phần giả lập đó bằng một API HTTP có phân trang và một PostgreSQL chứa dữ liệu bán hàng. Bạn sẽ có toàn bộ mã nguồn để chạy tại máy, không cần API key hoặc tài khoản dịch vụ ngoài.

Mục tiêu không dừng ở việc “gọi API ra JSON”. Pipeline phải đọc đủ trang, từ chối dữ liệu không hợp lệ, ghi theo một transaction và chạy lại mà không nhân đôi đơn. Chúng ta sẽ cố tình làm API trả lỗi và sửa trạng thái một đơn để kiểm tra hành vi thay vì chỉ nhìn một run màu xanh.

Tiếp tục thư mục ~/airflow-retail và stack Airflow 3.1.7 của bài 1. Giữ nguyên DAG retail_hello làm ví dụ riêng. Bài này thêm retail_api_to_postgres chạy tay; lịch theo ngày và backfill sẽ được đưa vào bài 3. Đơn vị tiền là VND nguyên, một dòng ứng với một đơn hàng; không phải một dòng sản phẩm trong đơn.

1. Hợp đồng dữ liệu trước khi viết pipeline

Một API không có hợp đồng rõ khiến code dễ “chạy được nhưng sai”. Trong lab, mỗi đơn có order_id là số nguyên dương, created_at có múi giờ, status thuộc paid/pending/cancelled và amount_vnd là số nguyên không âm. Doanh thu của ví dụ chỉ cộng đơn paid; tổng tiền các đơn pending hay cancelled không được gọi là doanh thu đã thanh toán.

Kết quả toàn bộ fixture phải có 5 đơn, 3 đơn paid và tổng paid 350.000 VND. Tổng giá trị mọi trạng thái là 470.000 VND. API trả tối đa 2 đơn một trang trong lệnh của pipeline, nên năm đơn tương ứng ba trang: 2, 2 và 1. Đây là các con số đối soát độc lập để phát hiện bỏ trang hoặc lọc trạng thái sai.

Chúng ta coi raw.orders là bảng trạng thái hiện tại của mỗi đơn. Nếu đơn 1005 đổi pending thành paid, cùng order_id được cập nhật chứ không thêm một bản sao. Lab chưa lưu mọi phiên bản thay đổi, chưa xử lý sự kiện xóa nguồn và chưa đưa ra bảo đảm exactly-once giữa hệ thống HTTP với PostgreSQL.

2. Thêm hai service, giữ database nghiệp vụ tách khỏi metadata Airflow

Trong thư mục project, tạo thư mục mock-api và file compose.override.yaml. Vì file gốc tên docker-compose.yaml, các lệnh dưới đây luôn truyền rõ cả hai -f, tránh phụ thuộc Compose có tự tìm tên override này trong môi trường của bạn hay không.

cd ~/airflow-retail
mkdir -p mock-api

services:
  shop-db:
    image: postgres:16.6
    environment:
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: shop_lab_only
      POSTGRES_DB: retail
    ports:
      - "127.0.0.1:55433:5432"
    volumes:
      - shop-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U shop -d retail"]
      interval: 5s
      timeout: 5s
      retries: 10
    restart: unless-stopped
  mock-api:
    image: python:3.12-slim
    working_dir: /app
    command: ["python", "server.py"]
    volumes:
      - ./mock-api:/app:ro
    ports:
      - "127.0.0.1:8088:8000"
    restart: unless-stopped
  airflow-worker:
    environment:
      AIRFLOW_CONN_SHOP_POSTGRES: "postgresql://shop:shop_lab_only@shop-db:5432/retail"
      SHOP_API_URL: "http://mock-api:8000/orders"
  airflow-apiserver:
    environment:
      AIRFLOW_CONN_SHOP_POSTGRES: "postgresql://shop:shop_lab_only@shop-db:5432/retail"
volumes:
  shop-data:

shop-db là PostgreSQL chứa đơn hàng. Service postgres có sẵn trong file gốc vẫn giữ metadata Airflow. Tách hai nơi giúp không nhầm bảng trạng thái task với bảng doanh thu, đồng thời tránh cho truy vấn báo cáo tác động trực tiếp database điều phối. Mật khẩu shop_lab_only là credential công khai chỉ dành cho môi trường học tại máy; khi triển khai thật dùng cơ chế quản lý secrets và quyền phù hợp.

mock-api dùng Python standard library, không phải cài thêm framework. Thư mục mock-api trên host được mount read-only vào /app trong container. Read-only nghĩa là tiến trình trong container không sửa file tại đó; bạn vẫn có thể sửa fixture trên host. API lắng nghe 0.0.0.0 bên trong container để service khác kết nối, nhưng cổng publish ra host chỉ bind 127.0.0.1.

Hai service Airflow được bổ sung environment, không thay thế toàn bộ environment kế thừa từ file gốc: worker cần URL API và connection PostgreSQL; API server cũng có connection cùng ID để nhất quán khi Airflow xử lý truy xuất cấu hình trong luồng thực thi. Lệnh config ở phần sau cho phép kiểm tra kết quả merge trước khi khởi động.

3. Tạo API giả lập: fixture cố định, phân trang và bộ lọc thời gian

Tạo mock-api/orders.json. Giữ đúng timestamp có +07:00 để bài 3 dùng lại cho khoảng ngày tại Việt Nam:

[
  {"order_id":1001,"created_at":"2026-01-01T09:00:00+07:00","status":"paid","amount_vnd":120000},
  {"order_id":1002,"created_at":"2026-01-01T14:00:00+07:00","status":"paid","amount_vnd":80000},
  {"order_id":1003,"created_at":"2026-01-01T18:00:00+07:00","status":"cancelled","amount_vnd":50000},
  {"order_id":1004,"created_at":"2026-01-02T10:00:00+07:00","status":"paid","amount_vnd":150000},
  {"order_id":1005,"created_at":"2026-01-02T22:00:00+07:00","status":"pending","amount_vnd":70000}
]

Tạo mock-api/server.py:

import json
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

ROOT = Path(__file__).parent

def aware(value):
    result = datetime.fromisoformat(value)
    if result.tzinfo is None or result.utcoffset() is None:
        raise ValueError('Timestamp must have timezone')
    return result

class Handler(BaseHTTPRequestHandler):
    def send_json(self, status, payload):
        body = json.dumps(payload).encode('utf-8')
        self.send_response(status)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        request = urlparse(self.path)
        if request.path != '/orders':
            self.send_json(404, {'error': 'not found'})
            return
        if (ROOT / 'fail.flag').exists():
            self.send_json(503, {'error': 'intentional lab outage'})
            return
        try:
            args = parse_qs(request.query)
            page = int(args.get('page', ['1'])[0])
            limit = int(args.get('limit', ['2'])[0])
            if page < 1 or not 1 <= limit <= 100:
                raise ValueError('Invalid page or limit')
            rows = json.loads((ROOT / 'orders.json').read_text())
            start, end = args.get('from'), args.get('to')
            if bool(start) != bool(end):
                raise ValueError('Provide both from and to')
            if start:
                lower, upper = aware(start[0]), aware(end[0])
                if lower >= upper:
                    raise ValueError('Invalid interval')
                rows = [r for r in rows
                        if lower <= aware(r['created_at']) < upper]
            rows.sort(key=lambda r: r['order_id'])
            offset = (page - 1) * limit
            self.send_json(200, {
                'items': rows[offset:offset + limit],
                'next_page': page + 1 if offset + limit < len(rows) else None,
            })
        except (ValueError, KeyError, TypeError) as exc:
            self.send_json(400, {'error': str(exc)})

if __name__ == '__main__':
    ThreadingHTTPServer(('0.0.0.0', 8000), Handler).serve_forever()

Endpoint /orders nhận page, limit và tùy chọn from/to. Khi chưa có from/to, nó trả toàn bộ fixture theo order_id; khi có cả hai, nó lọc khoảng thời gian đầu đóng, cuối mở: lấy created_at lớn hơn hoặc bằng from và nhỏ hơn to. Một đơn đúng tại mốc kết thúc sẽ thuộc khoảng kế tiếp. Server luôn trả object gồm items và next_page; next_page=null đánh dấu hết trang.

API đọc lại orders.json mỗi request để tiện thử thay đổi dữ liệu. Vì vậy fixture phải được giữ ổn định trong lúc một batch đang chạy. Trong API thật có ghi đồng thời, phân trang theo offset/page có thể bỏ hoặc lặp bản ghi khi nguồn thay đổi. Cần contract phù hợp như cursor ổn định, snapshot token hoặc watermark; ví dụ nhỏ này không tự giải quyết tính nhất quán của một API bất kỳ.

File fail.flag là công tắc lỗi có chủ đích. Khi tồn tại, endpoint trả HTTP 503 để bạn quan sát retry. Mã này không thực hiện xác thực, phân quyền hay rate limiting, nên chỉ dùng cho lab nội bộ tại máy.

4. Khởi động và phân biệt localhost với tên service

docker compose -f docker-compose.yaml -f compose.override.yaml config --quiet
docker compose -f docker-compose.yaml -f compose.override.yaml up -d \
  shop-db mock-api airflow-apiserver airflow-worker

docker compose -f docker-compose.yaml -f compose.override.yaml ps
curl --fail 'http://localhost:8088/orders?page=1&limit=2'

Lệnh curl chạy trên host dùng localhost:8088. Khi task chạy trong airflow-worker, localhost lại là chính container worker, nên task phải gọi http://mock-api:8000/orders. Tương tự, worker kết nối shop-db:5432; công cụ SQL trên host dùng localhost:55433. Đây là khác biệt về vị trí chương trình chạy, không phải tùy hứng chọn port nào cũng được.

Kiểm tra image Airflow đang có dependency cần dùng. Bản regular image của lab bao gồm các provider thông dụng; nếu bạn dùng slim/custom image, phải xây image bổ sung dependency theo constraints tương ứng, không cài package mỗi lần task chạy:

docker compose exec airflow-worker python -c \
  "import requests, psycopg2; from airflow.providers.postgres.hooks.postgres import PostgresHook; print('Dependencies OK')"

AIRFLOW_CONN_SHOP_POSTGRES định nghĩa connection ID shop_postgres qua environment. Connection kiểu này có thể không xuất hiện như một dòng được tạo trong trang Connections của UI. Không thêm một connection cùng ID nhưng khác host rồi kỳ vọng Airflow tự chọn đúng. Nếu đổi environment trong file Compose, cần tái tạo service liên quan bằng up -d để container nhận giá trị mới.

5. DAG đầy đủ: fetch, validate rồi upsert trong một transaction

Tạo dags/retail_common.py để chứa hai hàm validate_order và ingest_orders dùng chung. File này không tạo DAG, tránh tự đăng ký DAG ngoài ý muốn khi module được import ở bài 3. Ở top-level chỉ định nghĩa hàm và cấu trúc DAG; gọi HTTP, lấy connection và ghi database đều nằm trong phần thực thi task.

from datetime import datetime


def validate_order(row):
    if type(row.get('order_id')) is not int or row['order_id'] <= 0:
        raise ValueError('order_id must be a positive integer')
    if row.get('status') not in {'paid', 'pending', 'cancelled'}:
        raise ValueError(f"Unsupported status for order {row['order_id']}")
    amount = row.get('amount_vnd')
    if type(amount) is not int or not 0 <= amount <= 10**15:
        raise ValueError('amount_vnd must be an integer between 0 and 10^15')
    created_at = datetime.fromisoformat(row['created_at'])
    if created_at.tzinfo is None or created_at.utcoffset() is None:
        raise ValueError('created_at must have timezone')
    return (row['order_id'], created_at, row['status'], amount)


def ingest_orders(start=None, end=None):
    import os
    import requests
    from psycopg2.extras import execute_values
    from airflow.providers.postgres.hooks.postgres import PostgresHook

    filters = {}
    if (start is None) != (end is None):
        raise ValueError('Provide both interval boundaries')
    if start is not None:
        filters = {'from': start, 'to': end}
    page, seen_pages, orders = 1, set(), {}
    with requests.Session() as session:
        while page is not None:
            if type(page) is not int or page < 1 or page in seen_pages:
                raise ValueError('Invalid or repeated next_page')
            if len(seen_pages) >= 100:
                raise ValueError('Too many API pages for this lab')
            seen_pages.add(page)
            response = session.get(
                os.environ['SHOP_API_URL'],
                params={**filters, 'page': page, 'limit': 2},
                timeout=(3.05, 10),
            )
            response.raise_for_status()
            payload = response.json()
            if not isinstance(payload, dict):
                raise ValueError('Expected JSON object')
            if not isinstance(payload.get('items'), list) or 'next_page' not in payload:
                raise ValueError('Invalid pagination payload')
            for row in payload['items']:
                if not isinstance(row, dict):
                    raise ValueError('Expected order object')
                record = validate_order(row)
                if record[0] in orders:
                    raise ValueError('Duplicate order across API pages')
                orders[record[0]] = record
                if len(orders) > 10000:
                    raise ValueError('Batch too large for this lab')
            page = payload['next_page']

    connection = PostgresHook(postgres_conn_id='shop_postgres').get_conn()
    try:
        with connection:
            with connection.cursor() as cursor:
                cursor.execute('create schema if not exists raw')
                cursor.execute('''
                    create table if not exists 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
                    )
                ''')
                if orders:
                    execute_values(cursor, '''
                        insert into raw.orders
                            (order_id, created_at, status, amount_vnd)
                        values %s
                        on conflict (order_id) do update set
                            created_at = excluded.created_at,
                            status = excluded.status,
                            amount_vnd = excluded.amount_vnd,
                            loaded_at = current_timestamp
                        where (orders.created_at, orders.status, orders.amount_vnd)
                            is distinct from
                            (excluded.created_at, excluded.status, excluded.amount_vnd)
                    ''', list(orders.values()), page_size=1000)
        return {'fetched_orders': len(orders), 'pages': len(seen_pages),
                'from': start, 'to': end}
    finally:
        connection.close()

Tiếp theo tạo dags/retail_ingest.py. File này chỉ khai báo DAG và import hàm xử lý từ module dùng chung:

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


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

    @task
    def audit(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(*), coalesce(sum(amount_vnd)
                        filter (where status = 'paid'), 0)
                    from raw.orders
                ''')
                count, paid_value = cursor.fetchone()
            print({'batch': batch, 'total_orders': count,
                   'paid_value_vnd': int(paid_value)})
            if batch['fetched_orders'] == 0:
                raise ValueError('This full-scan lab expects nonempty input')
        finally:
            connection.close()

    audit(load_batch())

retail_api_to_postgres()

load_batch gọi ingest_orders và chỉ trả metadata: số đơn đã lấy, số trang, phạm vi lọc. audit nhận metadata đó rồi truy vấn database để in tổng hiện có. Audit trong bài này quan sát toàn bảng vì DAG chạy full scan; nó không khẳng định tổng toàn bảng luôn bằng số đơn của một batch sau khi bạn thêm dữ liệu khác.

Việc giữ extract và load trong cùng một task là lựa chọn có chủ ý cho fixture nhỏ. Không có file tạm trong container mà task sau buộc phải tìm thấy, và không truyền toàn bộ dữ liệu qua XCom. Khi batch lớn, cần thiết kế landing trên object storage, manifest theo batch và các bước độc lập; đừng chỉ tăng giới hạn 10.000 rồi hy vọng worker chịu được mọi khối lượng.

6. Những phần quyết định tính đúng của pipeline

6.1. Phân trang không được kết thúc bằng phỏng đoán

Vòng lặp chỉ dừng khi next_page là null theo contract API. seen_pages phát hiện trang bị lặp, giới hạn 100 trang ngăn vòng lặp vô hạn, và giới hạn batch tránh vô tình nạp quá nhiều vào RAM. Dữ liệu trùng order_id giữa các trang bị từ chối để không âm thầm chọn một bản ghi khi nguồn không nhất quán.

timeout=(3.05, 10) là timeout kết nối và timeout đọc của requests, không phải tổng thời gian cho toàn job. execution_timeout 5 phút là một giới hạn khác ở task. raise_for_status khiến 4xx/5xx trở thành exception thay vì cố coi trang HTML lỗi là JSON đơn hàng. Trường next_page bắt buộc tồn tại; thiếu nó phải báo sai contract, không được coi là đã hết dữ liệu.

6.2. Kiểm tra ở Python và database có mục đích bổ sung

Python từ chối amount dạng chuỗi, bool hoặc số thực để tránh nhập nhằng đơn vị tiền. type(value) is int cố ý không chấp nhận True vì bool là subclass của int trong Python. Timestamp phải có timezone để không âm thầm diễn giải theo múi giờ của container. Database vẫn có primary key, not null và check constraint để bảo vệ dữ liệu khi có writer khác.

Một lỗi validation trong code hiện tại vẫn chịu chính sách retries=1 của load_batch. Retry không sửa được một status lạ; lab giữ cấu hình đơn giản để quan sát, còn phân loại lỗi dữ liệu không retry và lỗi hạ tầng có retry sẽ được khai thác sâu ở bài 5. Không bắt exception rồi chỉ print “failed” và return thành công.

6.3. Transaction giữ batch ghi vào database theo một đơn vị

Code tải và kiểm tra xong mọi trang trước khi mở connection ghi. Trong with connection, tạo schema/bảng và toàn bộ các lệnh upsert nằm trong transaction của connection. execute_values có thể chia nhiều câu INSERT theo page_size=1000, nhưng các câu đó vẫn dùng chung transaction. Nếu lỗi trước commit, dữ liệu ghi trong transaction này được rollback; không có chuyện trang đầu đã commit riêng rồi trang sau mới phát hiện lỗi.

Giới hạn này chỉ nói về database của task load. API và PostgreSQL không nằm trong một distributed transaction. Ngoài ra audit là task sau, chạy sau khi load đã commit: nếu audit lỗi, dữ liệu của load không tự bị rollback. Trạng thái failed của cả DAG không đồng nghĩa tất cả tác động của các task trước đã được hoàn tác.

6.4. Idempotency là kết quả quan sát được khi chạy lại

Primary key order_id và ON CONFLICT DO UPDATE giúp cùng một đơn không sinh dòng thứ hai. excluded là bản ghi đang được đề nghị insert; orders là bản ghi hiện tại trong bảng đích. Điều kiện IS DISTINCT FROM chỉ cập nhật khi created_at, status hoặc amount thay đổi, nên chạy lại payload giống nhau không làm loaded_at nhảy liên tục.

loaded_at ở đây là thời điểm bản ghi được ghi mới hoặc thay đổi trong warehouse, không phải bằng chứng ingestion đã hoàn thành một batch mới. Nếu cần heartbeat, hãy lưu trạng thái batch riêng. Điều kiện khác biệt không ngăn một bản cập nhật cũ ghi đè bản mới; API thật cần updated_at/version và chính sách xử lý stale updates. max_active_runs=1 của DAG cũng không khóa writer thuộc một DAG khác.

7. Chạy lần đầu rồi đối soát bằng SQL

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

Chờ DAG processor nhận file, mở run trong UI và xem log audit. Với database mới và fixture gốc, metadata batch phải báo fetched_orders=5, pages=3; tổng bảng có 5 đơn và paid_value_vnd=350000. Nếu không thấy DAG, kiểm tra import errors trước khi suy đoán lỗi connection.

docker compose -f docker-compose.yaml -f compose.override.yaml exec -T shop-db \
  psql -U shop -d retail -c "
    select count(*) as orders,
           count(*) filter (where status='paid') as paid_orders,
           coalesce(sum(amount_vnd) filter (where status='paid'),0) as paid_vnd
    from raw.orders;"

Trigger lần thứ hai khi fixture chưa đổi. Kết quả phải vẫn là 5 dòng và 350.000 VND, không thành 10 dòng. Đây là một kiểm tra chạy lại có ý nghĩa hơn việc chỉ thấy task success hai lần. Có thể so loaded_at của từng order trước và sau để xác nhận bản ghi không đổi không bị update vô ích.

8. Thử ba tình huống quan trọng

8.1. API tạm thời không sẵn sàng

touch mock-api/fail.flag

Trigger run mới. load_batch phải lỗi HTTP 503 và có thể chuyển up_for_retry; audit chưa được chạy vì upstream chưa đạt. Xóa đúng file lỗi do bạn tạo trước lượt retry hoặc trước run kế tiếp:

rm mock-api/fail.flag

Lệnh rm này chỉ xóa công tắc thử nghiệm, không xóa database. Nếu bỏ cờ lỗi đủ sớm, lượt retry có thể thành công; nếu task đã hết lượt và failed, tạo run mới để kiểm tra khôi phục. Số dòng warehouse phải không tăng vì lần tải thất bại chưa bước vào transaction ghi.

8.2. Dữ liệu có status hoặc tiền không hợp lệ

Tạm sửa amount_vnd của đơn 1002 thành -1 trong fixture, chờ request mới và Trigger. Task phải dừng ở validation. Kiểm tra raw.orders vẫn giữ số liệu từ lần thành công trước. Sau đó khôi phục 80000. Để tránh đọc dở file khi sửa, thực hiện thay đổi lúc không có run hoạt động; JSON sai cú pháp cũng làm API trả lỗi và task phải thất bại rõ ràng.

8.3. Đơn cũ chuyển trạng thái

Đổi status của đơn 1005 từ pending thành paid rồi chạy lại. Kỳ vọng vẫn 5 dòng, số paid tăng từ 3 lên 4 và paid_vnd tăng từ 350.000 lên 420.000. Điều này kiểm tra UPDATE của upsert, không chỉ nhánh INSERT. Sau khi thử, khôi phục pending và chạy lại để trở về fixture gốc trước bài 3.

Trong thực tế, một đơn bị xóa hoàn toàn khỏi API sẽ vẫn còn trong raw.orders của lab vì upsert không có thao tác delete. Không được tự suy ra “không xuất hiện trong batch” nghĩa là phải xóa: có thể đó là API lọc theo ngày, phân quyền hoặc phân trang thiếu. Muốn đồng bộ xóa phải có contract riêng như soft delete, tombstone hay snapshot đầy đủ được xác nhận.

9. Bài lab nghiệm thu và ranh giới áp dụng

1. Từ database mới, đọc đủ ba trang và đối soát 5 đơn, 3 paid, 350.000 VND.
2. Chạy lại fixture không đổi, xác nhận không tăng số dòng và không cập nhật loaded_at vô ích.
3. Tạo lỗi 503, tìm exception đầu tiên và kiểm tra audit không chạy khi load thất bại.
4. Tạo tiền âm, xác nhận warehouse giữ trạng thái từ lần thành công trước; khôi phục dữ liệu sau thử nghiệm.
5. Đổi 1005 sang paid, kiểm tra upsert và doanh thu 420.000; hoàn nguyên rồi chạy lại.
6. Giải thích được kết nối trên host khác trong worker, lý do không dùng metadata DB để lưu đơn và vì sao không truyền batch lớn qua XCom.

Pipeline này là một nền tảng nhỏ, có giới hạn cố ý: full scan, dữ liệu giữ trong RAM, nguồn ổn định trong lúc phân trang và một writer được quản lý. Nó chưa có CDC, lịch sử thay đổi, xử lý xóa hay bảo đảm snapshot nguồn. Bài 3 sẽ giữ cùng API và bảng đích, nhưng mỗi run chỉ lấy một khoảng ngày cụ thể để chạy theo lịch và chạy bù có thể kiểm chứng.

Nếu cần ôn phần kiến trúc và Docker, xem bài 1: Airflow từ đầu và DAG bán hàng đầu tiên.

Tài liệu tham khảo

• Airflow best practices: task và chạy lại
• PostgresHook
• Connection qua environment
• PostgreSQL INSERT và ON CONFLICT
• psycopg2 transaction control
• requests: timeouts