Apache Airflow 2.10資料管道實戰：5種DAG編排模式打造生產級ETL流水線
AI与大数据
ETL腳本失控、任務依賴混亂、資料管道靜默失敗
凌晨2點，資料倉庫的日增量從50萬除降到3萬，Airflow DAG顯示全綠通過——沒人知道上游API改了分頁邏輯；ETL腳本散落在20個Cron Job裡，上游失敗下游照跑，髒資料一路灌入報表；新來的同事寫了個DAG，循環依賴導致Airflow Scheduler直接卡死。2026年，Apache Airflow 2.10帶來了Dynamic Task Mapping增強、改進的TaskFlow API和更強大的資料感知調度——從DAG設計到生產監控，一套體系全搞定。
本文將從5種DAG編排模式出發，帶你完成TaskFlow API設計→動態任務映射並行ETL→自訂Operator與Sensor→XCom資料傳遞與分支→錯誤重試與SLA監控的全鏈路實戰，每一步都有完整可執行的Python程式碼。
Airflow核心概念
| 概念 | 說明 |
|---|---|
| DAG | 有向無環圖，定義任務之間的依賴關係和執行順序，是Airflow的核心抽象 |
| Task | DAG中的單個執行單元，對應一個Operator實例 |
| Operator | Task的執行邏輯封裝，如PythonOperator、BashOperator、自訂Operator |
| TaskFlow API | Airflow 2.0+引入的Python函數式DAG定義方式，用@task裝飾器替代Operator |
| XCom | 跨任務資料傳遞機制，預設支援48KB，2.10版本支援自訂XCom Backend |
| Dynamic Task Mapping | Airflow 2.3+引入的動態任務展開，根據執行時資料動態生成並行任務 |
| Sensor | 特殊的Operator，持續偵測條件是否滿足，滿足後下游任務才開始 |
| Connection | Airflow管理的連接資訊（資料庫、API等），避免在DAG中硬編碼密碼 |
| Pool | 資源池，限制同時執行的任務數量，防止資源過載 |
| Trigger | Airflow 2.2+引入的異步觸發器，讓Sensor不再佔用Worker Slot |
| SLA | Service Level Agreement，任務必須在指定時間內完成，逾時觸發告警 |
| Dataset | Airflow 2.4+引入的資料感知調度，生產者產出資料後消費者自動觸發 |
問題分析：資料管道編排的5大挑戰
- DAG設計複雜度失控：ETL任務從3個增長到30個，DAG檔案變成麵條程式碼
- 並行ETL手動編排：每天處理100個資料源，手動寫100個Task不現實
- 自訂邏輯難以復用：公司內部系統的API呼叫、資料驗證邏輯在每個DAG裡重複實現
- 跨任務資料傳遞受限：XCom預設48KB限制，大資料量傳遞報錯
- 生產環境可靠性不足：任務失敗無人重試，SLA逾時無人告警
分步實操：5種Airflow DAG編排模式
模式1：TaskFlow API設計DAG
# dags/taskflow_etl_dag.py
from datetime import datetime, timedelta
from airflow.decorators import dag, task
default_args = {
"owner": "data-team",
"retries": 3,
"retry_delay": timedelta(minutes=5),
}
@dag(
dag_id="taskflow_etl_pipeline",
default_args=default_args,
description="TaskFlow API ETL pipeline",
schedule="0 2 * * *",
start_date=datetime(2026, 1, 1),
catchup=False,
tags=["etl", "taskflow"],
)
def taskflow_etl_pipeline():
@task
def extract_user_events(execution_date: str) -> dict:
"""从PostgreSQL抽取用户事件数据"""
import psycopg2
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("postgres_source")
pg_conn = psycopg2.connect(
host=conn.host, port=conn.port, database=conn.schema,
user=conn.login, password=conn.password,
)
cur = pg_conn.cursor()
cur.execute(
"SELECT user_id, event_type, amount, created_at "
"FROM user_events "
"WHERE created_at >= %s AND created_at < %s",
(execution_date, execution_date + timedelta(days=1)),
)
rows = cur.fetchall()
pg_conn.close()
return {"row_count": len(rows), "data": [dict(zip(["user_id", "event_type", "amount", "created_at"], r)) for r in rows[:1000]]}
@task
def extract_api_data(execution_date: str) -> dict:
"""从REST API抽取数据"""
import requests
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("api_source")
resp = requests.get(
f"{conn.host}/v2/data",
headers={"Authorization": f"Bearer {conn.password}"},
params={"date": execution_date},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
return {"row_count": len(data.get("records", [])), "data": data.get("records", [])[:1000]}
@task
def transform(user_events: dict, api_data: dict) -> dict:
"""合并、去重、清洗数据"""
import pandas as pd
df_events = pd.DataFrame(user_events["data"])
df_api = pd.DataFrame(api_data["data"])
if df_events.empty and df_api.empty:
return {"row_count": 0, "data": []}
df_all = pd.concat([df_events, df_api], ignore_index=True)
df_all = df_all.drop_duplicates(subset=["user_id", "event_type"])
df_all = df_all.dropna(subset=["user_id"])
df_all["amount"] = df_all["amount"].fillna(0).astype(float)
return {"row_count": len(df_all), "data": df_all.to_dict("records")[:500]}
@task
def load(transformed: dict) -> int:
"""加载数据到数据仓库"""
import psycopg2
from airflow.hooks.base import BaseHook
if transformed["row_count"] == 0:
return 0
conn = BaseHook.get_connection("postgres_warehouse")
pg_conn = psycopg2.connect(
host=conn.host, port=conn.port, database=conn.schema,
user=conn.login, password=conn.password,
)
cur = pg_conn.cursor()
insert_sql = """
INSERT INTO analytics.user_events_daily
(user_id, event_type, amount, event_date)
VALUES (%s, %s, %s, %s)
ON CONFLICT (user_id, event_type, event_date) DO UPDATE
SET amount = EXCLUDED.amount
"""
batch = [(r.get("user_id"), r.get("event_type"), r.get("amount", 0), r.get("created_at")) for r in transformed["data"]]
cur.executemany(insert_sql, batch)
pg_conn.commit()
pg_conn.close()
return len(batch)
@task
def quality_check(loaded_count: int, transformed_count: int) -> str:
if transformed_count > 0 and loaded_count / transformed_count < 0.95:
raise ValueError(f"Quality gate failed: loaded {loaded_count}/{transformed_count}")
return "PASSED"
user_events = extract_user_events("{{ ds }}")
api_data = extract_api_data("{{ ds }}")
transformed = transform(user_events, api_data)
loaded = load(transformed)
quality_check(loaded, transformed["row_count"])
taskflow_etl_pipeline()
pip install "apache-airflow[postgres,amazon]==2.10.3"
airflow db init
airflow dags list | grep taskflow
airflow dags test taskflow_etl_pipeline 2026-06-19
模式2：Dynamic Task Mapping並行ETL
# dags/dynamic_mapping_dag.py
from datetime import datetime
from airflow.decorators import dag, task
@dag(dag_id="dynamic_mapping_etl", schedule="0 3 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "dynamic-mapping"])
def dynamic_mapping_etl():
@task
def discover_data_sources() -> list[dict]:
"""动态发现需要处理的数据源列表"""
import psycopg2
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("postgres_source")
pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
cur = pg_conn.cursor()
cur.execute("SELECT source_id, source_type, connection_id FROM data_sources WHERE active = true")
sources = [{"source_id": row[0], "source_type": row[1], "connection_id": row[2]} for row in cur.fetchall()]
pg_conn.close()
return sources
@task
def extract_from_source(source: dict) -> dict:
"""从单个数据源抽取数据"""
import psycopg2, requests
from airflow.hooks.base import BaseHook
source_type, source_id = source["source_type"], source["source_id"]
if source_type == "postgres":
conn = BaseHook.get_connection(source["connection_id"])
pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
cur = pg_conn.cursor()
cur.execute(f"SELECT * FROM source_{source_id}_data WHERE processed = false LIMIT 5000")
rows = cur.fetchall()
pg_conn.close()
return {"source_id": source_id, "row_count": len(rows), "status": "success"}
elif source_type == "api":
conn = BaseHook.get_connection(source["connection_id"])
resp = requests.get(f"{conn.host}/v1/data/{source_id}", headers={"Authorization": f"Bearer {conn.password}"}, timeout=30)
resp.raise_for_status()
return {"source_id": source_id, "row_count": len(resp.json()), "status": "success"}
return {"source_id": source_id, "row_count": 0, "status": "skipped"}
@task
def validate_extracted(extract_result: dict) -> dict:
if extract_result["status"] != "success" or extract_result["row_count"] < 1:
return {**extract_result, "valid": False}
return {**extract_result, "valid": True}
@task
def aggregate_results(validation_results: list[dict]) -> dict:
total = len(validation_results)
valid = sum(1 for r in validation_results if r.get("valid"))
total_rows = sum(r.get("row_count", 0) for r in validation_results if r.get("valid"))
return {"total_sources": total, "valid_sources": valid, "total_rows": total_rows}
# Dynamic Task Mapping: expand()将列表展开为并行任务
sources = discover_data_sources()
extract_results = extract_from_source.expand(source=sources)
validation_results = validate_extracted.expand(extract_result=extract_results)
aggregate_results(validation_results)
dynamic_mapping_etl()
# plugins/operators/internal_api_operator.py
from airflow.models import BaseOperator
from airflow.exceptions import AirflowSkipException
from typing import Any
import requests
class InternalApiOperator(BaseOperator):
"""调用公司内部API的自定义Operator"""
template_fields = ("endpoint", "payload")
def __init__(self, endpoint: str, payload: dict | None = None, method: str = "POST",
connection_id: str = "internal_api", timeout: int = 60, skip_on_empty: bool = False, **kwargs):
super().__init__(**kwargs)
self.endpoint, self.payload, self.method = endpoint, payload or {}, method
self.connection_id, self.timeout, self.skip_on_empty = connection_id, timeout, skip_on_empty
def execute(self, context: Any) -> dict:
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection(self.connection_id)
url = f"{conn.host}{self.endpoint}"
headers = {"Authorization": f"Bearer {conn.password}", "Content-Type": "application/json"}
self.log.info(f"Calling {self.method} {url}")
try:
if self.method.upper() == "GET":
resp = requests.get(url, headers=headers, params=self.payload, timeout=self.timeout)
else:
resp = requests.post(url, headers=headers, json=self.payload, timeout=self.timeout)
resp.raise_for_status()
result = resp.json()
if self.skip_on_empty and not result:
raise AirflowSkipException("API returned empty result")
return result
except requests.exceptions.Timeout:
raise TimeoutError(f"API call timed out after {self.timeout}s")
except requests.exceptions.HTTPError as e:/n if e.response.status_code == 404 and self.skip_on_empty:
raise AirflowSkipException(f"Resource not found: {url}")
raise
# plugins/sensors/data_ready_sensor.py
from airflow.sensors.base import BaseSensorOperator
from typing import Any
class DataReadySensor(BaseSensorOperator):
"""检测数据源是否已就绪的Sensor"""
template_fields = ("source_table", "expected_date")
def __init__(self, source_table: str, expected_date: str, connection_id: str = "postgres_source",
min_row_count: int = 1, poke_interval: int = 300, timeout: int = 7200, **kwargs):
super().__init__(poke_interval=poke_interval, timeout=timeout, **kwargs)
self.source_table, self.expected_date = source_table, expected_date
self.connection_id, self.min_row_count = connection_id, min_row_count
def poke(self, context: Any) -> bool:
import psycopg2
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection(self.connection_id)
pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
cur = pg_conn.cursor()
cur.execute(f"SELECT COUNT(*) FROM {self.source_table} WHERE partition_date = %s AND status = 'ready'", (self.expected_date,))
count = cur.fetchone()[0]
pg_conn.close()
self.log.info(f"Data ready check: {count} rows (min: {self.min_row_count})")
return count >= self.min_row_count
模式3：自訂Operator和Sensor
# dags/xcom_branching_dag.py
from datetime import datetime
from airflow.decorators import dag, task
from airflow.operators.python import BranchPythonOperator
@dag(dag_id="xcom_branching_pipeline", schedule="0 5 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "xcom", "branching"])
def xcom_branching_pipeline():
@task
def detect_data_type() -> dict:
import psycopg2
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("postgres_source")
pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
cur = pg_conn.cursor()
cur.execute("SELECT COUNT(*), SUM(amount) FROM user_events WHERE created_at >= CURRENT_DATE - INTERVAL '1 day'")
row_count, total_amount = cur.fetchone()
pg_conn.close()
return {"row_count": row_count, "total_amount": float(total_amount or 0),
"data_size": "large" if row_count > 100000 else "small"}
@task
def process_small_batch(metadata: dict) -> dict:
return {"path": "small_batch", "processed": metadata["row_count"], "status": "completed"}
@task
def process_large_batch(metadata: dict) -> dict:
chunk_size = 10000
chunks = (metadata["row_count"] + chunk_size - 1) // chunk_size
return {"path": "large_batch", "processed": metadata["row_count"], "chunks": chunks, "status": "completed"}
def choose_processing_branch(**context):
ti = context["ti"]
metadata = ti.xcom_pull(task_ids="detect_data_type")
return "process_large_batch" if metadata["data_size"] == "large" else "process_small_batch"
metadata = detect_data_type()
branch = BranchPythonOperator(task_id="choose_branch", python_callable=choose_processing_branch)
small_result = process_small_batch(metadata)
large_result = process_large_batch(metadata)
metadata >> branch >> [small_result, large_result]
xcom_branching_pipeline()
# custom_xcom_backend.py - 自定义XCom Backend突破48KB限制
import json, hashlib
from airflow.models.xcom import BaseXCom
from typing import Any
class RedisXComBackend(BaseXCom):
@staticmethod
def serialize_value(value: Any) -> bytes:
import redis
r = redis.Redis(host="redis-host", port=6379, db=2)
serialized = json.dumps(value, default=str).encode("utf-8")
if len(serialized) <= 48000:
return BaseXCom.serialize_value(value)
key = f"xcom:{hashlib.md5(serialized).hexdigest()}"
r.setex(key, 86400 * 7, serialized)
return BaseXCom.serialize_value({"__xcom_redis_key": key, "size": len(serialized)})
@staticmethod
def deserialize_value(result: Any) -> Any:
import redis
value = BaseXCom.deserialize_value(result)
if isinstance(value, dict) and "__xcom_redis_key" in value:
r = redis.Redis(host="redis-host", port=6379, db=2)
data = r.get(value["__xcom_redis_key"])
if data: return json.loads(data)
raise ValueError(f"XCom data expired in Redis")
return value
# airflow.cfg: xcom_backend = custom_xcom_backend.RedisXComBackend
模式4：XCom資料傳遞與管道分支
# dags/error_handling_sla_dag.py
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.exceptions import AirflowSkipException, AirflowFailOperator
default_args = {
"owner": "data-team", "retries": 3, "retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True, "max_retry_delay": timedelta(minutes=30),
"execution_timeout": timedelta(minutes=60),
"email_on_failure": True, "email": ["data-team@company.com"],
}
@dag(dag_id="error_handling_sla_pipeline", default_args=default_args,
schedule="0 1 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["etl", "production", "sla"])
def error_handling_sla_pipeline():
@task(retries=5, retry_delay=timedelta(minutes=2), retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=20))
def extract_with_retry(**context) -> dict:
import requests
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("api_source")
try:
resp = requests.get(f"{conn.host}/v2/data", headers={"Authorization": f"Bearer {conn.password}"}, params={"date": context["ds"]}, timeout=30)
resp.raise_for_status()
return {"row_count": len(resp.json().get("records", [])), "status": "success"}
except requests.exceptions.ConnectionError as e:/n raise ConnectionError(f"Connection failed: {e}")
except requests.exceptions.HTTPError as e:/n if e.response.status_code == 429: raise ConnectionError(f"Rate limited: {e}")
elif e.response.status_code >= 500: raise ConnectionError(f"Server error: {e}")
elif e.response.status_code == 404: raise AirflowSkipException(f"Data not found")
else: raise AirflowFailOperator(f"Client error: {e}")
@task
def transform_with_validation(raw_data: dict) -> dict:
import pandas as pd
if raw_data.get("status") != "success": raise ValueError("Upstream failed")
df = pd.DataFrame(raw_data.get("records", []))
if df.empty: raise AirflowSkipException("No data")
null_rate = df.isnull().sum().sum() / (df.shape[0] * df.shape[1])
if null_rate > 0.3: raise ValueError(f"Data quality too low: {null_rate:.1%}")
return {"row_count": len(df), "null_rate": null_rate, "status": "transformed"}
@task
def load_with_idempotency(transformed: dict) -> dict:
import psycopg2
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("postgres_warehouse")
pg_conn = psycopg2.connect(host=conn.host, port=conn.port, database=conn.schema, user=conn.login, password=conn.password)
cur = pg_conn.cursor()
cur.execute("DELETE FROM analytics.daily_metrics WHERE metric_date = CURRENT_DATE - INTERVAL '1 day'")
cur.execute("INSERT INTO analytics.daily_metrics (metric_date, row_count, null_rate, status) VALUES (CURRENT_DATE - INTERVAL '1 day', %s, %s, %s)", (transformed["row_count"], transformed["null_rate"], transformed["status"]))
pg_conn.commit(); pg_conn.close()
return {"loaded": True, "inserted": 1}
def sla_check(**context):
import pendulum
execution_date = context["execution_date"]
if pendulum.now("UTC") > execution_date.add(hours=2):
raise ValueError("SLA violated")
return "SLA_OK"
sla_check_task = PythonOperator(task_id="sla_check", python_callable=sla_check)
sla_alert_task = PythonOperator(task_id="sla_alert", python_callable=lambda **c: print(f"SLA Alert: {c['dag'].dag_id}"), trigger_rule="one_failed")
raw = extract_with_retry()
transformed = transform_with_validation(raw)
loaded = load_with_idempotency(transformed)
loaded >> sla_check_task >> sla_alert_task
error_handling_sla_pipeline()
# Airflow 2.10数据感知调度
from airflow.datasets import Dataset
USER_EVENTS = Dataset("postgres://warehouse/analytics/user_events_daily")
ORDER_METRICS = Dataset("postgres://warehouse/analytics/order_metrics_daily")
@dag(schedule="0 2 * * *")
def producer():
@task(outlets=[USER_EVENTS])
def produce_user_events(): return "produced"
@task(outlets=[ORDER_METRICS])
def produce_order_metrics(): return "produced"
produce_user_events(); produce_order_metrics()
@dag(schedule=[USER_EVENTS, ORDER_METRICS]) # 两个Dataset都更新后触发
def consumer():
@task
def generate_daily_report(): return "report_generated"
generate_daily_report()
模式5：錯誤處理、重試與SLA監控
# ❌ 每次Scheduler解析都会执行查询
sources = requests.get("https://api.example.com/sources").json()
@dag(...)
def my_dag():
@task
def process():
for source in sources: ...
# ✅ 动态逻辑放在Task内部
@dag(...)
def my_dag():
@task
def get_sources() -> list:
return requests.get("https://api.example.com/sources").json()
@task
def process(sources: list):
for source in sources: ...
5大避坑指南
1. DAG檔案頂層程式碼執行資料庫查詢
❌ 錯誤做法：在DAG檔案頂層執行資料庫查詢或API呼叫，每次Scheduler解析都會執行
✅ 正確做法：將動態邏輯放在Task內部，DAG檔案頂層只做結構定義
# ❌ DataFrame通过XCom传递
@task
def extract():
df = pd.read_csv("huge_file.csv")
return df.to_dict() # XCom序列化失败
2. XCom傳遞大資料物件
❌ 錯誤做法：透過XCom傳遞整個DataFrame，超過48KB直接報錯
✅ 正確做法：傳遞檔案路徑引用，或使用自訂XCom Backend
# ✅ 传递文件路径引用
@task
def extract():
df = pd.read_csv("huge_file.csv")
path = f"/tmp/extract_{datetime.now().strftime('%Y%m%d')}.parquet"
df.to_parquet(path)
return {"file_path": path, "row_count": len(df)}
@task
def transform(metadata: dict):
df = pd.read_parquet(metadata["file_path"])
# 处理逻辑...
3. catchup=True導致歷史回填雪崩
❌ 錯誤做法：DAG設定catchup=True且start_date在一年前，啟動後疯狂回填365個DAG Run
✅ 正確做法：生產DAG一律catchup=False，需要回填時用airflow dags backfill手動控制
4. Sensor佔用Worker Slot導致死鎗
❌ 錯誤做法：多個Sensor同時執行，每個佔用一個Worker Slot
✅ 正確做法：使用mode="reschedule"或deferrable模式釋放Worker Slot
# ✅ reschedule模式
FileSensor(task_id="wait", filepath="/data/{{ ds }}.csv", mode="reschedule")
# ✅ deferrable模式（Airflow 2.2+推荐）
FileSensor(task_id="wait", filepath="/data/{{ ds }}.csv", deferrable=True)
5. 忽略資料品質門禁
❌ 錯誤做法：DAG只檢查任務是否執行成功，空資料也顯示綠色
✅ 正確做法：在DAG末尾加品質門禁，資料量或品質不達標則阻斷
@task
def quality_gate(loaded_count: int, expected_min: int = 1000) -> str:
if loaded_count < expected_min:
raise ValueError(f"Quality gate failed: {loaded_count} < {expected_min}")
return "PASSED"
10大報錯排查
| 序號 | 報錯訊息 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | DAG import error |
Worker環境缺少Python依賴套件 | 在requirements.txt中新增依賴，重啟Worker |
| 2 | XCom value size exceeded 48KB |
XCom傳遞資料超過預設48KB限制 | 傳遞檔案路徑引用；或配置自訂XCom Backend（Redis） |
| 3 | Pool has no free slots |
資源池已滿，任務排隊等待 | 增大Pool的slot數量：airflow pool set default_pool 128 |
| 4 | DAG seems missing from DagBag |
DAG檔案有語法錯誤或循環匯入 | 執行airflow dags list-import-errors檢查 |
| 5 | Sensor timeout |
Sensor等待條件始終不滿足 | 檢查資料源；增大timeout；使用deferrable模式 |
| 6 | AirflowTaskTimeout |
Task執行時間超過execution_timeout |
增大timeout；拆分為子任務；優化程式碼效能 |
| 7 | Detected deadlock |
DAG存在循環依賴或資源競爭 | 檢查DAG依賴圖；檢查Pool配置 |
| 8 | Connection refused |
資料庫或API連接失敗 | 檢查Connection配置：airflow connections get-uri |
| 9 | Dynamic task mapping too many tasks |
展開的任務數量超過max_map_length |
調整airflow.cfg中max_map_length（預設1024） |
| 10 | SLA missed |
任務未在SLA規定時間內完成 | 檢查上游延遲；優化效能；調整SLA時間窗口 |
進階優化技巧
1. DAG解析效能優化
@dag(
dag_id="optimized_pipeline",
dagrun_timeout=timedelta(hours=3), # DAG Run最多3小时
max_active_runs=1, # 同时只允许1个DAG Run
max_active_tasks=16, # 最多16个并发Task
)
def optimized_pipeline(): ...
2. 連接池與批量操作
from airflow.hooks.base import BaseHook
import psycopg2
from contextlib import contextmanager
@contextmanager
def get_db_connection(connection_id: str):
conn_info = BaseHook.get_connection(connection_id)
conn = psycopg2.connect(host=conn_info.host, port=conn_info.port, database=conn_info.schema, user=conn_info.login, password=conn_info.password)
try: yield conn
finally: conn.close()
def batch_insert(conn, table: str, columns: list[str], rows: list[tuple], batch_size: int = 5000):
from psycopg2.extras import execute_values
with conn.cursor() as cur:
for i in range(0, len(rows), batch_size):
execute_values(cur, f"INSERT INTO {table} ({', '.join(columns)}) VALUES %s", rows[i:i + batch_size])
conn.commit()
3. Dataset資料感知調度替代Cron
from airflow.datasets import Dataset
RAW_DATA = Dataset("s3://data-lake/raw/events/")
CLEANED_DATA = Dataset("s3://data-lake/cleaned/events/")
@dag(schedule="0 2 * * *")
def producer():
@task(outlets=[RAW_DATA])
def extract(): ...
@task(outlets=[CLEANED_DATA])
def clean(): ...
@dag(schedule=[CLEANED_DATA]) # 数据就绪后自动触发
def consumer():
@task
def train_model(): ...
4. 優先順序與資源隔離
# 关键DAG使用高优先级和专用Pool
@dag(dag_id="critical_pipeline", priority_weight=100, pool="revenue_pool")
def critical_pipeline(): ...
# 非关键DAG使用低优先级
@dag(dag_id="experimental_pipeline", priority_weight=1, pool="default_pool")
def experimental_pipeline(): ...
5. 監控指標與告警整合
# airflow.cfg 开启StatsD指标导出
# [metrics]
# statsd_on = True
# statsd_host = localhost
# statsd_port = 8125
def sla_callback(dag, task_list, blocking_task_list, dagrun, *args, **kwargs):
import requests
from airflow.models import Variable
webhook_url = Variable.get("slack_webhook_url")
message = {"text": f":warning: SLA Violation\nDAG: {dag.dag_id}\nRun: {dagrun.execution_date}\nBlocking: {[t.task_id for t in blocking_task_list]}"}
requests.post(webhook_url, json=message)
default_args = {"sla": timedelta(hours=2), "sla_miss_callback": sla_callback}
編排工具對比
| 維度 | Apache Airflow | Prefect | Dagster | Luigi |
|---|---|---|---|---|
| 編排模式 | DAG定義式 | 函數式+裝飾器 | 資產定義式 | 任務依賴式 |
| 資料感知調度 | Dataset（2.4+） | 原生支援 | 原生Asset | 不支援 |
| 動態任務 | Dynamic Task Mapping | 原生動態 | 原生動態 | 不支援 |
| 自訂Operator | 強大，插件生態豐富 | 裝飾器即可 | Software-defined Asset | 簡單Task類 |
| XCom/資料傳遞 | 48KB限制，可自訂Backend | 原生無限制 | 原生Type系統 | 簡單參數傳遞 |
| 錯誤處理 | 重試+SLA+回呼 | 原生重試+狀態 | 原生重試+Asset狀態 | 簡單重試 |
| 監控 | Web UI + StatsD | Cloud UI | Dagit UI | 基礎Web |
| 學習曲線 | 中等 | 低 | 中高 | 低 |
| 社區生態 | 最成熟，插件最多 | 快速增長 | 增長中 | 衰退 |
| Kubernetes | KubernetesPodOperator | 原生K8s Agent | 原生K8s | 不支援 |
| 2026推薦度 | ★★★★★ | ★★★★ | ★★★★ | ★★ |
| 適用場景 | 企業級ETL編排 | 中小專案快速啟動 | 資料資產密集型 | 簡單批次處理 |
相關工具
- JSON格式化 — 除錯Airflow DAG組態和API回應資料時格式化JSON
- 雜湊加密 — 產生XCom Redis Backend的資料指紋和驗證資料完整性
- cURL轉程式碼 — 將API除錯的cURL命令轉為Python程式碼用於ETL擷取任務
延伸閱讀
- Python AI資料流水線實戰 — 從ETL到特徵儲存的完整資料流水線
- Python Celery分佈式任務佇列 — 分佈式任務編排的另一種選擇
- Python Pydantic V2資料驗證 — 資料管道中的Schema驗證防線
- Apache Kafka流處理 — 即時資料管道的流處理方案
Apache Airflow 2.10在2026年依然是資料管道編排的事實標準。記住五個核心原則：TaskFlow API是DAG設計的首選——函數式風格比Operator更簡潔可讀；Dynamic Task Mapping是並行ETL的利器——一個Task定義展開為N個實例；自訂Operator是復用的基石——內部系統邏輯封裝一次，全團隊受益；XCom+分支是靈活調度的關鍵——資料驗動決策；錯誤處理和SLA是生產的底線——沒有重試和監控的DAG，上線就是定時炸彈。
本站提供瀏覽器本地工具，免註冊即可試用 →
#Airflow数据管道#ETL编排#DAG工作流#Airflow 2.10#2026#AI与大数据