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 给出了一个完整的入门 DAGdags/example_dag.pyfrom 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(minutes5), retry_exponential_backoff: True, max_retry_delay: timedelta(hours1), } with DAG( dag_idexample_etl, default_argsdefault_args, descriptionExample ETL pipeline, schedule0 6 * * *, # Daily at 6 AM start_datedatetime(2024, 1, 1), catchupFalse, tags[etl, example], max_active_runs1, ) as dag: start EmptyOperator(task_idstart) def extract_data(**context): execution_date context[ds] # Extract logic here return {records: 1000} extract PythonOperator( task_idextract, python_callableextract_data, ) end EmptyOperator(task_idend) start extract enddefault_args 参数详解default_args中的配置会被组内所有 Task 继承Task 自身显式声明的同名参数优先级更高理解每个键是写出可靠重试策略的前提参数作用与建议owner负责人标识便于告警与追溯归属depends_on_past是否依赖上一调度周期成功生产中默认False置True会形成串联瓶颈email_on_failure任务失败时是否发邮件告警email_on_retry重试时是否发邮件通常False避免告警风暴retries失败后的自动重试次数如 3retry_delay每次重试前的等待间隔如 5 分钟retry_exponential_backoff是否指数退避等待时间随重试次数指数增长降低对下游系统的冲击max_retry_delay指数退避的等待上限如 1 小时防止退避时间无限拉长DAG 级参数说明dag_id全局唯一标识会出现在 UI、日志、告警与 API 中。schedule调度表达式。支持 cron0 6 * * *表示每天 06:00与预设daily、hourly、weekly等。Airflow 2.4 推荐用schedule替代旧的schedule_interval。start_dateDAG 的起始日期配合catchup决定补跑行为。catchupFalse不追溯补跑启动前错过的周期盲目开启追跑backfill在数据量大时会瞬时打爆调度队列。tags给 DAG 打标签便于 UI 筛选如etl、example。max_active_runs1同一时间只允许一个 DAG Run 在跑避免上一周期未结束、下一周期又启动造成的并发写冲突。任务上下文与dspython_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 APIAirflow 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_idtaskflow_etl, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, 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(fs3://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(fs3://bucket/{target}/{{ ds }}.parquet) return transformed[rows] task() def notify(rows_loaded: int): Send notification print(fLoaded {rows_loaded} rows) # Define dependencies with XCom passing extracted extract(sourceraw_data) transformed transform(extracted) loaded load(transformed, targetprocessed_data) notify(loaded) # Instantiate the DAG taskflow_etl()使用要点依赖即“函数调用”transform(extracted)就同时完成了数据传递与依赖声明notify(loaded)前必须完成loadXCom 的序列化/反序列化全部自动完成。文件路径中{{ ds }}必须写成f-string双花括号转义fs3://.../{{ 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 fetl_{config[name]} default_args { owner: data-team, retries: 3, retry_delay: timedelta(minutes5), } dag DAG( dag_iddag_id, default_argsdefault_args, scheduleconfig[schedule], start_datedatetime(2024, 1, 1), catchupFalse, tags[etl, dynamic, config[name]], ) with dag: def extract_fn(source, **context): print(fExtracting from {source} for {context[ds]}) def transform_fn(**context): print(fTransforming data for {context[ds]}) def load_fn(table_name, **context): print(fLoading to {table_name} for {context[ds]}) extract PythonOperator( task_idextract, python_callableextract_fn, op_kwargs{source: config[source]}, ) transform PythonOperator( task_idtransform, python_callabletransform_fn, ) load PythonOperator( task_idload, python_callableload_fn, op_kwargs{table_name: config[name]}, ) extract transform load return dag # Generate DAGs for config in PIPELINE_CONFIGS: globals()[fdag_{config[name]}] create_dag(config)要点与陷阱每个 config 会生成一个独立 DAGdag_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_idbranching_pipeline, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, ) 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_idscheck_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_idbranch, python_callablechoose_branch, ) high_quality EmptyOperator(task_idhigh_quality_path) medium_quality EmptyOperator(task_idmedium_quality_path) low_quality EmptyOperator(task_idlow_quality_path) # Join point - runs after any branch completes join EmptyOperator( task_idjoin, trigger_ruleTriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) quality_check branch [high_quality, medium_quality, low_quality] join branching_pipeline()运行机制拆解check_data_qualityTaskFlow返回质量分自动写入 XCom。choose_branch通过context[ti].xcom_pull(task_idscheck_data_quality)读回指标并返回分支任务 id本例设了三个阈值档位0.9、0.7、其余分别对应high/medium/low_quality_path。分支各自是一个独立任务未被选中的分支任务会被标记为 skipped。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_idsensor_example, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: # Wait for file on S3 wait_for_file S3KeySensor( task_idwait_for_s3_file, bucket_namedata-lake, bucket_keyraw/{{ ds }}/data.parquet, aws_conn_idaws_default, timeout60 * 60 * 2, # 2 hours poke_interval60 * 5, # Check every 5 minutes modereschedule, # Free up worker slot while waiting ) # Wait for another DAG to complete wait_for_upstream ExternalTaskSensor( task_idwait_for_upstream_dag, external_dag_idupstream_etl, external_task_idfinal_task, execution_date_fnlambda dt: dt, # Same execution date timeout60 * 60 * 3, modereschedule, ) # Custom sensor using task.sensor decorator task.sensor(poke_interval60, timeout3600, modereschedule) 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_doneis_done, xcom_valueresponse.json()) api_ready wait_for_api() def process_data(**context): api_result context[ti].xcom_pull(task_idswait_for_api) print(fAPI returned: {api_result}) process PythonOperator( task_idprocess, python_callableprocess_data, ) [wait_for_file, wait_for_upstream, api_ready] process关键参数与选择依据场景使用关键参数数据文件到达S3/HDFS/本地S3KeySensorAWS 需安装apache-airflow-providers-amazon、FileSensorbucket_name、bucket_key支持{{ ds }}宏、aws_conn_id等待另一个 DAG 的某个 TaskExternalTaskSensorexternal_dag_id、external_task_id、execution_date_fn对齐执行日期自定义轮询逻辑API 健康等task.sensor装饰器poke_interval、timeout、mode轮询频率poke_interval默认约 60 秒控制每次探测间隔这里 S3 探测设为 5 分钟timeout为最长等待超时任务即失败本例分别为 2 小时与 3 小时。modereschedule是生产关键与默认的poke模式在 Task 槽位内死等不同reschedule模式在两次探测之间会释放 worker 槽位避免长时间等待的 Sensor 占满执行器——这正是 SKILL.md Best Practices 中“sensor 使用modereschedule以释放 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_idswait_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_iderror_handling_example, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, on_failure_callbackdag_failure_callback, default_args{ on_failure_callback: task_failure_callback, retries: 3, retry_delay: timedelta(minutes5), }, ) as dag: def might_fail(**context): import random if random.random() 0.3: raise ValueError(Random failure!) return Success risky_task PythonOperator( task_idrisky_task, python_callablemight_fail, ) def cleanup(**context): Cleanup runs regardless of upstream failures print(Cleaning up...) cleanup_task PythonOperator( task_idcleanup, python_callablecleanup, trigger_ruleTriggerRule.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_idnotify_success, python_callablenotify_success, trigger_ruleTriggerRule.ALL_SUCCESS, ) risky_task [cleanup_task, success_notification]结构分析两级回调分工on_failure_callback经default_args注入所有 Task负责单任务粒度的失败告警组装dag_id、task_id、ds、异常对象与log_url发送到 Slack/PagerDutydag_failure_callback在 DAG 级汇总失败并生成摘要。回调收到的context是标准执行上下文context[task_instance].log_url可直接定位失败日志。“无论成败都清理”cleanup_task用TriggerRule.ALL_DONErisky_task 无论成功还是抛异常都会触发清理适合释放锁、关闭连接等收尾动作。“全成功才通知”success_notification用默认的TriggerRule.ALL_SUCCESS只有 risky_task 成功才发送成功通知。该示例还与 SKILL.md 的重试体系串联retries: 3retry_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_folderdags/, include_examplesFalse) def test_dag_loaded(dagbag): Test that all DAGs load without errors assert len(dagbag.import_errors) 0, fDAG 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, fCycle 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(ds2024-01-01) assert records in result assert isinstance(result[records], int)测试方法拆解DagBag解析测试DagBag(dag_folderdags/, include_examplesFalse)模拟调度器解析目录import_errors非空即说明有 DAG 无法被解析通常是语法错、导入错、模板渲染错必须在 CI 中拦截。结构与依赖测试用dag.get_task(extract)取出任务再断言其upstream_list/downstream_list符合预期start → extract → end防止重构时悄悄改坏拓扑。无环完整性测试遍历dagbag.dags逐个执行dag.test_cycle()任何环都会导致调度异常。该测试对应前文 任务依赖 一节中复杂依赖图的正确性保障。纯函数单元测试把extract_data这类可独立执行的函数抽取成普通函数后直接单测示例中通过ds2024-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 PracticesDos 与 Donts 清单SKILL.md 的收尾部分浓缩了实战中血泪经验这里结合模式示例逐条展开应该做Dos使用 TaskFlow API代码更简洁XCom 自动传递见模式一。这是 Airflow 2.x 的主推写法。设置超时timeouts为任务设置执行超时与传感器timeout防止“僵尸任务”长期占用 worker长轮询场景务必配modereschedule释放槽位。Sensor 使用modereschedule避免大量等待型 Sensor 阻塞并发池见模式四。测试 DAG用 DagBag 做结构/无环/依赖断言配合纯函数单测见模式六。保证任务幂等重试与补跑才能安全与 DAG 四大原则的Idempotent对应。不要做Donts不要轻易用depends_on_pastTrue会让某个周期失败后下游周期全部积压成瓶颈如需串行控制优先用max_active_runs1。不要硬编码日期用{{ ds }}宏与执行上下文派生否则回填、跨时区、历史重跑都会出错。不要使用全局状态Task 应无状态共享可变变量在多 worker 部署下行为不可预期模式二为此专门用工厂隔离配置。不要盲目跳过 catchupcatchupFalse是常见默认但要理解其语义——需要补历史数据时需显式 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Apache Airflow DAG 生产级设计模式实战指南：基于 agents 数据工程技能构建可靠的编排管道
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 给出了一个完整的入门 DAGdags/example_dag.pyfrom 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(minutes5), retry_exponential_backoff: True, max_retry_delay: timedelta(hours1), } with DAG( dag_idexample_etl, default_argsdefault_args, descriptionExample ETL pipeline, schedule0 6 * * *, # Daily at 6 AM start_datedatetime(2024, 1, 1), catchupFalse, tags[etl, example], max_active_runs1, ) as dag: start EmptyOperator(task_idstart) def extract_data(**context): execution_date context[ds] # Extract logic here return {records: 1000} extract PythonOperator( task_idextract, python_callableextract_data, ) end EmptyOperator(task_idend) start extract enddefault_args 参数详解default_args中的配置会被组内所有 Task 继承Task 自身显式声明的同名参数优先级更高理解每个键是写出可靠重试策略的前提参数作用与建议owner负责人标识便于告警与追溯归属depends_on_past是否依赖上一调度周期成功生产中默认False置True会形成串联瓶颈email_on_failure任务失败时是否发邮件告警email_on_retry重试时是否发邮件通常False避免告警风暴retries失败后的自动重试次数如 3retry_delay每次重试前的等待间隔如 5 分钟retry_exponential_backoff是否指数退避等待时间随重试次数指数增长降低对下游系统的冲击max_retry_delay指数退避的等待上限如 1 小时防止退避时间无限拉长DAG 级参数说明dag_id全局唯一标识会出现在 UI、日志、告警与 API 中。schedule调度表达式。支持 cron0 6 * * *表示每天 06:00与预设daily、hourly、weekly等。Airflow 2.4 推荐用schedule替代旧的schedule_interval。start_dateDAG 的起始日期配合catchup决定补跑行为。catchupFalse不追溯补跑启动前错过的周期盲目开启追跑backfill在数据量大时会瞬时打爆调度队列。tags给 DAG 打标签便于 UI 筛选如etl、example。max_active_runs1同一时间只允许一个 DAG Run 在跑避免上一周期未结束、下一周期又启动造成的并发写冲突。任务上下文与dspython_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 APIAirflow 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_idtaskflow_etl, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, 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(fs3://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(fs3://bucket/{target}/{{ ds }}.parquet) return transformed[rows] task() def notify(rows_loaded: int): Send notification print(fLoaded {rows_loaded} rows) # Define dependencies with XCom passing extracted extract(sourceraw_data) transformed transform(extracted) loaded load(transformed, targetprocessed_data) notify(loaded) # Instantiate the DAG taskflow_etl()使用要点依赖即“函数调用”transform(extracted)就同时完成了数据传递与依赖声明notify(loaded)前必须完成loadXCom 的序列化/反序列化全部自动完成。文件路径中{{ ds }}必须写成f-string双花括号转义fs3://.../{{ 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 fetl_{config[name]} default_args { owner: data-team, retries: 3, retry_delay: timedelta(minutes5), } dag DAG( dag_iddag_id, default_argsdefault_args, scheduleconfig[schedule], start_datedatetime(2024, 1, 1), catchupFalse, tags[etl, dynamic, config[name]], ) with dag: def extract_fn(source, **context): print(fExtracting from {source} for {context[ds]}) def transform_fn(**context): print(fTransforming data for {context[ds]}) def load_fn(table_name, **context): print(fLoading to {table_name} for {context[ds]}) extract PythonOperator( task_idextract, python_callableextract_fn, op_kwargs{source: config[source]}, ) transform PythonOperator( task_idtransform, python_callabletransform_fn, ) load PythonOperator( task_idload, python_callableload_fn, op_kwargs{table_name: config[name]}, ) extract transform load return dag # Generate DAGs for config in PIPELINE_CONFIGS: globals()[fdag_{config[name]}] create_dag(config)要点与陷阱每个 config 会生成一个独立 DAGdag_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_idbranching_pipeline, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, ) 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_idscheck_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_idbranch, python_callablechoose_branch, ) high_quality EmptyOperator(task_idhigh_quality_path) medium_quality EmptyOperator(task_idmedium_quality_path) low_quality EmptyOperator(task_idlow_quality_path) # Join point - runs after any branch completes join EmptyOperator( task_idjoin, trigger_ruleTriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) quality_check branch [high_quality, medium_quality, low_quality] join branching_pipeline()运行机制拆解check_data_qualityTaskFlow返回质量分自动写入 XCom。choose_branch通过context[ti].xcom_pull(task_idscheck_data_quality)读回指标并返回分支任务 id本例设了三个阈值档位0.9、0.7、其余分别对应high/medium/low_quality_path。分支各自是一个独立任务未被选中的分支任务会被标记为 skipped。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_idsensor_example, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: # Wait for file on S3 wait_for_file S3KeySensor( task_idwait_for_s3_file, bucket_namedata-lake, bucket_keyraw/{{ ds }}/data.parquet, aws_conn_idaws_default, timeout60 * 60 * 2, # 2 hours poke_interval60 * 5, # Check every 5 minutes modereschedule, # Free up worker slot while waiting ) # Wait for another DAG to complete wait_for_upstream ExternalTaskSensor( task_idwait_for_upstream_dag, external_dag_idupstream_etl, external_task_idfinal_task, execution_date_fnlambda dt: dt, # Same execution date timeout60 * 60 * 3, modereschedule, ) # Custom sensor using task.sensor decorator task.sensor(poke_interval60, timeout3600, modereschedule) 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_doneis_done, xcom_valueresponse.json()) api_ready wait_for_api() def process_data(**context): api_result context[ti].xcom_pull(task_idswait_for_api) print(fAPI returned: {api_result}) process PythonOperator( task_idprocess, python_callableprocess_data, ) [wait_for_file, wait_for_upstream, api_ready] process关键参数与选择依据场景使用关键参数数据文件到达S3/HDFS/本地S3KeySensorAWS 需安装apache-airflow-providers-amazon、FileSensorbucket_name、bucket_key支持{{ ds }}宏、aws_conn_id等待另一个 DAG 的某个 TaskExternalTaskSensorexternal_dag_id、external_task_id、execution_date_fn对齐执行日期自定义轮询逻辑API 健康等task.sensor装饰器poke_interval、timeout、mode轮询频率poke_interval默认约 60 秒控制每次探测间隔这里 S3 探测设为 5 分钟timeout为最长等待超时任务即失败本例分别为 2 小时与 3 小时。modereschedule是生产关键与默认的poke模式在 Task 槽位内死等不同reschedule模式在两次探测之间会释放 worker 槽位避免长时间等待的 Sensor 占满执行器——这正是 SKILL.md Best Practices 中“sensor 使用modereschedule以释放 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_idswait_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_iderror_handling_example, scheduledaily, start_datedatetime(2024, 1, 1), catchupFalse, on_failure_callbackdag_failure_callback, default_args{ on_failure_callback: task_failure_callback, retries: 3, retry_delay: timedelta(minutes5), }, ) as dag: def might_fail(**context): import random if random.random() 0.3: raise ValueError(Random failure!) return Success risky_task PythonOperator( task_idrisky_task, python_callablemight_fail, ) def cleanup(**context): Cleanup runs regardless of upstream failures print(Cleaning up...) cleanup_task PythonOperator( task_idcleanup, python_callablecleanup, trigger_ruleTriggerRule.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_idnotify_success, python_callablenotify_success, trigger_ruleTriggerRule.ALL_SUCCESS, ) risky_task [cleanup_task, success_notification]结构分析两级回调分工on_failure_callback经default_args注入所有 Task负责单任务粒度的失败告警组装dag_id、task_id、ds、异常对象与log_url发送到 Slack/PagerDutydag_failure_callback在 DAG 级汇总失败并生成摘要。回调收到的context是标准执行上下文context[task_instance].log_url可直接定位失败日志。“无论成败都清理”cleanup_task用TriggerRule.ALL_DONErisky_task 无论成功还是抛异常都会触发清理适合释放锁、关闭连接等收尾动作。“全成功才通知”success_notification用默认的TriggerRule.ALL_SUCCESS只有 risky_task 成功才发送成功通知。该示例还与 SKILL.md 的重试体系串联retries: 3retry_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_folderdags/, include_examplesFalse) def test_dag_loaded(dagbag): Test that all DAGs load without errors assert len(dagbag.import_errors) 0, fDAG 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, fCycle 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(ds2024-01-01) assert records in result assert isinstance(result[records], int)测试方法拆解DagBag解析测试DagBag(dag_folderdags/, include_examplesFalse)模拟调度器解析目录import_errors非空即说明有 DAG 无法被解析通常是语法错、导入错、模板渲染错必须在 CI 中拦截。结构与依赖测试用dag.get_task(extract)取出任务再断言其upstream_list/downstream_list符合预期start → extract → end防止重构时悄悄改坏拓扑。无环完整性测试遍历dagbag.dags逐个执行dag.test_cycle()任何环都会导致调度异常。该测试对应前文 任务依赖 一节中复杂依赖图的正确性保障。纯函数单元测试把extract_data这类可独立执行的函数抽取成普通函数后直接单测示例中通过ds2024-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 PracticesDos 与 Donts 清单SKILL.md 的收尾部分浓缩了实战中血泪经验这里结合模式示例逐条展开应该做Dos使用 TaskFlow API代码更简洁XCom 自动传递见模式一。这是 Airflow 2.x 的主推写法。设置超时timeouts为任务设置执行超时与传感器timeout防止“僵尸任务”长期占用 worker长轮询场景务必配modereschedule释放槽位。Sensor 使用modereschedule避免大量等待型 Sensor 阻塞并发池见模式四。测试 DAG用 DagBag 做结构/无环/依赖断言配合纯函数单测见模式六。保证任务幂等重试与补跑才能安全与 DAG 四大原则的Idempotent对应。不要做Donts不要轻易用depends_on_pastTrue会让某个周期失败后下游周期全部积压成瓶颈如需串行控制优先用max_active_runs1。不要硬编码日期用{{ ds }}宏与执行上下文派生否则回填、跨时区、历史重跑都会出错。不要使用全局状态Task 应无状态共享可变变量在多 worker 部署下行为不可预期模式二为此专门用工厂隔离配置。不要盲目跳过 catchupcatchupFalse是常见默认但要理解其语义——需要补历史数据时需显式 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
