企业官网建设流程全解析

Apache Airflow DAG 生产级设计模式实战指南：基于 agents 数据工程技能构建可靠的编排管道

【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents

本指南围绕 agents 插件仓库中># Linear 线性 task1 >> task2 >> task3 # Fan-out 扇出：task1 完成后并行执行三个分支 task1 >> [task2, task3, task4] # Fan-in 汇聚：三个上游全部成功后执行 task4 [task1, task2, task3] >> task4 # Complex 复杂组合：task2、task3 可并行，task4 等待二者 task1 >> task2 >> task4 task1 >> task3 >> task4

理解要点：

• a >> b等价于a.set_downstream(b)，即“a 先于 b”；a << b为反向“b 先于 a”。
• 列表与单个 Operator 混用（如[task1, task2] >> task4）表示 fan-in；单个 Operator 接列表表示 fan-out。
• 分支内任务并行度由调度器依据依赖图与执行器并发上限决定，DAG 作者只需要声明“谁依赖谁”。
• 无论拓扑多复杂，Airflow 会按依赖关系计算执行顺序；这也是 details.md 中“测试 DAG 无环”的验证基础——dag.test_cycle()即用来检测依赖图中是否出现循环。

Quick Start：一个可直接运行的 ETL 示例

SKILL.md 给出了一个完整的入门 DAG（dags/example_dag.py）：

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.empty import EmptyOperator default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), 'retry_exponential_backoff': True, 'max_retry_delay': timedelta(hours=1), } with DAG( dag_id='example_etl', default_args=default_args, description='Example ETL pipeline', schedule='0 6 * * *', # Daily at 6 AM start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'example'], max_active_runs=1, ) as dag: start = EmptyOperator(task_id='start') def extract_data(**context): execution_date = context['ds'] # Extract logic here return {'records': 1000} extract = PythonOperator( task_id='extract', python_callable=extract_data, ) end = EmptyOperator(task_id='end') start >> extract >> end

default_args 参数详解

default_args中的配置会被组内所有 Task 继承（Task 自身显式声明的同名参数优先级更高），理解每个键是写出可靠重试策略的前提：

DAG 级参数说明

• dag_id：全局唯一标识，会出现在 UI、日志、告警与 API 中。
• schedule：调度表达式。支持 cron（'0 6 * * *'表示每天 06:00）与预设（@daily、@hourly、@weekly等）。Airflow 2.4+ 推荐用schedule替代旧的schedule_interval。
• start_date：DAG 的起始日期，配合catchup决定补跑行为。
• catchup=False：不追溯补跑启动前错过的周期；盲目开启追跑（backfill）在数据量大时会瞬时打爆调度队列。
• tags：给 DAG 打标签，便于 UI 筛选（如etl、example）。
• max_active_runs=1：同一时间只允许一个 DAG Run 在跑，避免上一周期未结束、下一周期又启动造成的并发写冲突。

任务上下文与ds

python_callable函数声明**context后即可拿到 Airflow 注入的运行上下文。示例中的context['ds']即“逻辑执行日期”的YYYY-MM-DD字符串，对应模板宏{{ ds }}。不要在 DAG 代码里硬编码日期，一律通过ds、ts、execution_date等上下文或模板宏派生，这既是增量处理的锚点，也是数据回溯（rerun）可复现的前提。返回值的字典（如{'records': 1000}）会被自动写入 XCom，供下游任务读取。

六大进阶模式与完整示例

当 SKILL.md 的导航层不足以覆盖场景时，应按 details.md 的指引读取深度示例。六个模式构成一条从“编码风格”到“生产运维”的完整能力链：TaskFlow API → 动态 DAG → 分支 → 传感器 → 错误处理与告警 → 测试。

模式一：TaskFlow API（Airflow 2.0+）

传统PythonOperator需要手写op_kwargs、xcom_push/xcom_pull，样板代码多。TaskFlow API 用@dag与@task装饰器把普通 Python 函数变成 DAG 与任务，函数返回值自动通过 XCom 在任务间传递，只需把上游任务的返回值当作参数传入下游函数即可：

# dags/taskflow_example.py from datetime import datetime from airflow.decorators import dag, task from airflow.models import Variable @dag( dag_id='taskflow_etl', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'taskflow'], ) def taskflow_etl(): """ETL pipeline using TaskFlow API""" @task() def extract(source: str) -> dict: """Extract data from source""" import pandas as pd df = pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv') return {'data': df.to_dict(), 'rows': len(df)} @task() def transform(extracted: dict) -> dict: """Transform extracted data""" import pandas as pd df = pd.DataFrame(extracted['data']) df['processed_at'] = datetime.now() df = df.dropna() return {'data': df.to_dict(), 'rows': len(df)} @task() def load(transformed: dict, target: str): """Load data to target""" import pandas as pd df = pd.DataFrame(transformed['data']) df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet') return transformed['rows'] @task() def notify(rows_loaded: int): """Send notification""" print(f'Loaded {rows_loaded} rows') # Define dependencies with XCom passing extracted = extract(source='raw_data') transformed = transform(extracted) loaded = load(transformed, target='processed_data') notify(loaded) # Instantiate the DAG taskflow_etl()

使用要点：

• 依赖即“函数调用”：transform(extracted)就同时完成了数据传递与依赖声明，notify(loaded)前必须完成load，XCom 的序列化/反序列化全部自动完成。
• 文件路径中{{ ds }}必须写成f-string双花括号转义（f's3://.../{{ ds }}.csv'），否则会在 DAG 解析期被提前渲染。
• 由于依赖来自函数参数，函数签名类型提示（-> dict、-> int）尽量写全，可读性与 IDE 支持更好。
• 该风格对应># dags/dynamic_dag_factory.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.models import Variable import json # Configuration for multiple similar pipelines PIPELINE_CONFIGS = [ {'name': 'customers', 'schedule': '@daily', 'source': 's3://raw/customers'}, {'name': 'orders', 'schedule': '@hourly', 'source': 's3://raw/orders'}, {'name': 'products', 'schedule': '@weekly', 'source': 's3://raw/products'}, ] def create_dag(config: dict) -> DAG: """Factory function to create DAGs from config""" dag_id = f"etl_{config['name']}" default_args = { 'owner': 'data-team', 'retries': 3, 'retry_delay': timedelta(minutes=5), } dag = DAG( dag_id=dag_id, default_args=default_args, schedule=config['schedule'], start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'dynamic', config['name']], ) with dag: def extract_fn(source, **context): print(f"Extracting from {source} for {context['ds']}") def transform_fn(**context): print(f"Transforming data for {context['ds']}") def load_fn(table_name, **context): print(f"Loading to {table_name} for {context['ds']}") extract = PythonOperator( task_id='extract', python_callable=extract_fn, op_kwargs={'source': config['source']}, ) transform = PythonOperator( task_id='transform', python_callable=transform_fn, ) load = PythonOperator( task_id='load', python_callable=load_fn, op_kwargs={'table_name': config['name']}, ) extract >> transform >> load return dag # Generate DAGs for config in PIPELINE_CONFIGS: globals()[f"dag_{config['name']}"] = create_dag(config)

  要点与陷阱：

  • 每个 config 会生成一个独立 DAG，dag_id如etl_customers、etl_orders、etl_products，调度频率互不影响（@daily/@hourly/@weekly）。
  • 不同管道差异化的参数（source、table_name）通过op_kwargs注入对应任务，闭包/工厂负责隔离，避免共享可变状态。
  • 关键陷阱：Airflow Scheduler 会周期性地重新解析整个 DAG 文件以发现变化。因此配置文件若写死为模块级常量（如示例的PIPELINE_CONFIGS），新增管道需重发代码；若配置来自外部源（数据库、Variable、文件），则可在调度重解析时动态增删 DAG，这正是该模式适合“配置驱动平台”的原因。本示例显式 import 了Variable与json，即暗示配置可演进为运行时来源。
  • globals()赋值必须在模块顶层执行，且要在文件末尾调用工厂，保证 DAG 对象在解析结束时已绑定。

  模式三：分支逻辑与 TriggerRule

  数据管道往往需要按数据特征走不同分支（如按质量分路由）。BranchPythonOperator返回下游任务 id决定走哪条路径；由于分支导致部分上游未执行，汇合点必须改用合适的TriggerRule，否则默认的all_success会让汇合任务永远失败：

  # dags/branching_example.py from airflow.decorators import dag, task from airflow.operators.python import BranchPythonOperator from airflow.operators.empty import EmptyOperator from airflow.utils.trigger_rule import TriggerRule @dag( dag_id='branching_pipeline', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) def branching_pipeline(): @task() def check_data_quality() -> dict: """Check data quality and return metrics""" quality_score = 0.95 # Simulated return {'score': quality_score, 'rows': 10000} def choose_branch(**context) -> str: """Determine which branch to execute""" ti = context['ti'] metrics = ti.xcom_pull(task_ids='check_data_quality') if metrics['score'] >= 0.9: return 'high_quality_path' elif metrics['score'] >= 0.7: return 'medium_quality_path' else: return 'low_quality_path' quality_check = check_data_quality() branch = BranchPythonOperator( task_id='branch', python_callable=choose_branch, ) high_quality = EmptyOperator(task_id='high_quality_path') medium_quality = EmptyOperator(task_id='medium_quality_path') low_quality = EmptyOperator(task_id='low_quality_path') # Join point - runs after any branch completes join = EmptyOperator( task_id='join', trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) quality_check >> branch >> [high_quality, medium_quality, low_quality] >> join branching_pipeline()

  运行机制拆解：

  1. check_data_quality（TaskFlow）返回质量分，自动写入 XCom。
  2. choose_branch通过context['ti'].xcom_pull(task_ids='check_data_quality')读回指标并返回分支任务 id；本例设了三个阈值档位：>=0.9、>=0.7、其余，分别对应high/medium/low_quality_path。
  3. 分支各自是一个独立任务；未被选中的分支任务会被标记为 skipped。
  4. join使用TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS（只要有至少一个上游成功、且没有失败即触发），确保无论走哪条分支都能继续，这正是分支汇合的经典写法。其它常见规则还包括ALL_DONE（无论成败都执行）、ALL_SUCCESS（全部成功才执行）、ONE_SUCCESS（任一成功即执行）。

  模式四：Sensors 与外部依赖

  Sensor 是“等待外部条件满足”的特殊任务。三类典型场景被组合在一个 DAG 中演示：等待 S3 文件就绪、等待上游 DAG 完成、轮询外部 API 健康状态：

  # dags/sensor_patterns.py from datetime import datetime, timedelta from airflow import DAG from airflow.sensors.filesystem import FileSensor from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.sensors.external_task import ExternalTaskSensor from airflow.operators.python import PythonOperator with DAG( dag_id='sensor_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) as dag: # Wait for file on S3 wait_for_file = S3KeySensor( task_id='wait_for_s3_file', bucket_name='data-lake', bucket_key='raw/{{ ds }}/data.parquet', aws_conn_id='aws_default', timeout=60 * 60 * 2, # 2 hours poke_interval=60 * 5, # Check every 5 minutes mode='reschedule', # Free up worker slot while waiting ) # Wait for another DAG to complete wait_for_upstream = ExternalTaskSensor( task_id='wait_for_upstream_dag', external_dag_id='upstream_etl', external_task_id='final_task', execution_date_fn=lambda dt: dt, # Same execution date timeout=60 * 60 * 3, mode='reschedule', ) # Custom sensor using @task.sensor decorator @task.sensor(poke_interval=60, timeout=3600, mode='reschedule') def wait_for_api() -> PokeReturnValue: """Custom sensor for API availability""" import requests response = requests.get('https://api.example.com/health') is_done = response.status_code == 200 return PokeReturnValue(is_done=is_done, xcom_value=response.json()) api_ready = wait_for_api() def process_data(**context): api_result = context['ti'].xcom_pull(task_ids='wait_for_api') print(f"API returned: {api_result}") process = PythonOperator( task_id='process', python_callable=process_data, ) [wait_for_file, wait_for_upstream, api_ready] >> process

  关键参数与选择依据：

  场景	使用	关键参数
  数据文件到达（S3/HDFS/本地）	S3KeySensor（AWS 需安装apache-airflow-providers-amazon）、FileSensor	bucket_name、bucket_key（支持{{ ds }}宏）、aws_conn_id
  等待另一个 DAG 的某个 Task	ExternalTaskSensor	external_dag_id、external_task_id、execution_date_fn（对齐执行日期）
  自定义轮询逻辑（API 健康等）	@task.sensor装饰器	poke_interval、timeout、mode

  • 轮询频率：poke_interval（默认约 60 秒）控制每次探测间隔，这里 S3 探测设为 5 分钟；timeout为最长等待，超时任务即失败，本例分别为 2 小时与 3 小时。
  • mode='reschedule'是生产关键：与默认的poke模式在 Task 槽位内死等不同，reschedule模式在两次探测之间会释放 worker 槽位，避免长时间等待的 Sensor 占满执行器——这正是 SKILL.md Best Practices 中“sensor 使用mode='reschedule'以释放 worker”的出处。注意reschedule模式对任务有额外约束（如超时后任务进入 deferred 状态并由 scheduler 重排）。
  • @task.sensor进阶能力：把自定义轮询逻辑写成装饰器函数，返回PokeReturnValue(is_done=..., xcom_value=...)，既做条件判断又向 XCom 写入探测结果（如 API 返回的 JSON）。示例为简洁起见省略了该类型的导入，实际使用需补from airflow.sensors.base import PokeReturnValue。
  • 三个传感器 fan-in 汇合到process，即“文件就绪 + 上游完成 + API 可达”三者全部满足才开始处理；process用context['ti'].xcom_pull(task_ids='wait_for_api')读回 API 结果。

  模式五：错误处理与告警

  生产 DAG 必须有“失败可感知、清理可执行、成功可通知”的能力。模式五同时演示了**回调（callback）与触发规则（TriggerRule）**两个机制：

  # dags/error_handling.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.utils.trigger_rule import TriggerRule from airflow.models import Variable def task_failure_callback(context): """Callback on task failure""" task_instance = context['task_instance'] exception = context.get('exception') # Send to Slack/PagerDuty/etc message = f""" Task Failed! DAG: {task_instance.dag_id} Task: {task_instance.task_id} Execution Date: {context['ds']} Error: {exception} Log URL: {task_instance.log_url} """ # send_slack_alert(message) print(message) def dag_failure_callback(context): """Callback on DAG failure""" # Aggregate failures, send summary pass with DAG( dag_id='error_handling_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, on_failure_callback=dag_failure_callback, default_args={ 'on_failure_callback': task_failure_callback, 'retries': 3, 'retry_delay': timedelta(minutes=5), }, ) as dag: def might_fail(**context): import random if random.random() < 0.3: raise ValueError("Random failure!") return "Success" risky_task = PythonOperator( task_id='risky_task', python_callable=might_fail, ) def cleanup(**context): """Cleanup runs regardless of upstream failures""" print("Cleaning up...") cleanup_task = PythonOperator( task_id='cleanup', python_callable=cleanup, trigger_rule=TriggerRule.ALL_DONE, # Run even if upstream fails ) def notify_success(**context): """Only runs if all upstream succeeded""" print("All tasks succeeded!") success_notification = PythonOperator( task_id='notify_success', python_callable=notify_success, trigger_rule=TriggerRule.ALL_SUCCESS, ) risky_task >> [cleanup_task, success_notification]

  结构分析：

  • 两级回调分工：on_failure_callback（经default_args注入所有 Task）负责单任务粒度的失败告警，组装dag_id、task_id、ds、异常对象与log_url发送到 Slack/PagerDuty；dag_failure_callback在 DAG 级汇总失败并生成摘要。回调收到的context是标准执行上下文，context['task_instance'].log_url可直接定位失败日志。
  • “无论成败都清理”：cleanup_task用TriggerRule.ALL_DONE，risky_task 无论成功还是抛异常都会触发清理，适合释放锁、关闭连接等收尾动作。
  • “全成功才通知”：success_notification用默认的TriggerRule.ALL_SUCCESS，只有 risky_task 成功才发送成功通知。
  • 该示例还与 SKILL.md 的重试体系串联：retries: 3+retry_delay: 5min保证瞬态失败自动恢复，callback 只在最终失败时报警，符合“可观测原则”的告警要求。

  模式六：测试 DAG

  “先写测试、再上生产”在 Airflow 中同样成立。details.md 给出了两类测试：结构测试（用DagBag验证 DAG 能被正确解析、依赖无环）与逻辑单元测试（直接调用 Python 函数验证业务逻辑）：

  # tests/test_dags.py import pytest from datetime import datetime from airflow.models import DagBag @pytest.fixture def dagbag(): return DagBag(dag_folder='dags/', include_examples=False) def test_dag_loaded(dagbag): """Test that all DAGs load without errors""" assert len(dagbag.import_errors) == 0, f"DAG import errors: {dagbag.import_errors}" def test_dag_structure(dagbag): """Test specific DAG structure""" dag = dagbag.get_dag('example_etl') assert dag is not None assert len(dag.tasks) == 3 assert dag.schedule_interval == '0 6 * * *' def test_task_dependencies(dagbag): """Test task dependencies are correct""" dag = dagbag.get_dag('example_etl') extract_task = dag.get_task('extract') assert 'start' in [t.task_id for t in extract_task.upstream_list] assert 'end' in [t.task_id for t in extract_task.downstream_list] def test_dag_integrity(dagbag): """Test DAG has no cycles and is valid""" for dag_id, dag in dagbag.dags.items(): assert dag.test_cycle() is None, f"Cycle detected in {dag_id}" # Test individual task logic def test_extract_function(): """Unit test for extract function""" from dags.example_dag import extract_data result = extract_data(ds='2024-01-01') assert 'records' in result assert isinstance(result['records'], int)

  测试方法拆解：

  • DagBag解析测试：DagBag(dag_folder='dags/', include_examples=False)模拟调度器解析目录；import_errors非空即说明有 DAG 无法被解析（通常是语法错、导入错、模板渲染错），必须在 CI 中拦截。
  • 结构与依赖测试：用dag.get_task('extract')取出任务，再断言其upstream_list/downstream_list符合预期（start → extract → end），防止重构时悄悄改坏拓扑。
  • 无环完整性测试：遍历dagbag.dags逐个执行dag.test_cycle()，任何环都会导致调度异常。该测试对应前文 任务依赖 一节中复杂依赖图的正确性保障。
  • 纯函数单元测试：把extract_data这类可独立执行的函数抽取成普通函数后直接单测（示例中通过ds='2024-01-01'显式注入执行日期），业务逻辑不必依赖真实 Airflow 运行时。
  • 一个注意点：test_dag_structure中断言的是旧属性dag.schedule_interval；在 Airflow 2.4+ 中用schedule声明调度后，测试可改断言dag.schedule，二者取决于你部署的 Airflow 版本。

  生产项目目录规范

  无论模式如何选，一个可维护的 Airflow 项目应遵循 details.md 给出的目录骨架：

  airflow/ ├── dags/ │ ├── __init__.py │ ├── common/ │ │ ├── __init__.py │ │ ├── operators.py # Custom operators │ │ ├── sensors.py # Custom sensors │ │ └── callbacks.py # Alert callbacks │ ├── etl/ │ │ ├── customers.py │ │ └── orders.py │ └── ml/ │ └── training.py ├── plugins/ │ └── custom_plugin.py ├── tests/ │ ├── __init__.py │ ├── test_dags.py │ └── test_operators.py ├── docker-compose.yml └── requirements.txt

  组织原则：

  • DAG 文件保持轻薄：SKILL.md 明确“不要把重逻辑塞进 DAG 文件”，dags/common/下集中放置自定义 Operator、Sensor 与告警回调，供多个 DAG 复用；etl/、ml/按业务域划分 DAG。
  • tests 与 dags 平行：测试文件按“解析 → 结构 → 无环 → 业务逻辑”分层，可在 CI 中用 pytest 直接跑通。
  • plugins/ 目录：存放 Airflow 插件（自定义 Hook、视图、宏等）；custom_plugin.py中的注册逻辑需与[plugins_folder]配置配合。
  • docker-compose.yml + requirements.txt：提供本地一键起 Airflow 与依赖锁定，让“本地可测”成为开发默认动作。

  Best Practices：Do's 与 Don'ts 清单

  SKILL.md 的收尾部分浓缩了实战中血泪经验，这里结合模式示例逐条展开：

  应该做（Do's）

  • 使用 TaskFlow API：代码更简洁，XCom 自动传递，见模式一。这是 Airflow 2.x 的主推写法。
  • 设置超时（timeouts）：为任务设置执行超时与传感器timeout，防止“僵尸任务”长期占用 worker；长轮询场景务必配mode='reschedule'释放槽位。
  • Sensor 使用mode='reschedule'：避免大量等待型 Sensor 阻塞并发池，见模式四。
  • 测试 DAG：用 DagBag 做结构/无环/依赖断言，配合纯函数单测，见模式六。
  • 保证任务幂等：重试与补跑才能安全，与 DAG 四大原则的Idempotent对应。

  不要做（Don'ts）

  • 不要轻易用depends_on_past=True：会让某个周期失败后下游周期全部积压成瓶颈；如需串行控制优先用max_active_runs=1。
  • 不要硬编码日期：用{{ ds }}宏与执行上下文派生，否则回填、跨时区、历史重跑都会出错。
  • 不要使用全局状态：Task 应无状态，共享可变变量在多 worker 部署下行为不可预期（模式二为此专门用工厂隔离配置）。
  • 不要盲目跳过 catchup：catchup=False是常见默认，但要理解其语义——需要补历史数据时需显式 backfill 或按需开启。
  • 不要把重逻辑放进 DAG 文件：DAG 文件会被调度器高频重复解析，内部应只做“组 DAG、拼依赖”，业务逻辑放到dags/common/模块或独立包中，见目录规范。

  总结与生态用法

  airflow-dag-patterns为 Airflow 生产实践提供了从“四原则 + 依赖语法 + 入门 DAG”到“TaskFlow / 动态 DAG / 分支 / Sensor / 告警 / 测试”的完整闭环。在 agents 仓库中，它通过渐进式披露（SKILL.md → references/details.md）控制上下文成本，并与 contenteditable="false">【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents