定时任务和事件驱动任务混在一起——触发方式设计

# 定时触发：每天凌晨1点
dag_daily = DAG('daily_etl', schedule='0 1 * * *', ...)
 
# 每小时触发
dag_hourly = DAG('hourly_metrics', schedule='0 * * * *', ...)
 
# 纯依赖触发（不定时，由外部Sensor或API触发）
dag_event = DAG('event_driven', schedule=None, ...)
 
# API触发示例（用curl触发DAG）
# curl -X POST http://airflow:8080/api/v1/dags/event_driven/dagRuns \
#   -H 'Content-Type: application/json' \
#   -d '{"conf": {"source": "ad_platform", "action": "budget_change"}}'

# 混合触发的DAG设计
 
# DAG 1：定时拉取（每小时）
with DAG('hourly_ad_pull', schedule='0 * * * *') as dag1:
    pull = PythonOperator(task_id='pull_ad_data', ...)
    process = PythonOperator(task_id='process_ad_data', ...)
    pull >> process
 
# DAG 2：事件驱动（API触发）
with DAG('event_budget_change', schedule=None) as dag2:
    # dag_run.conf里包含事件信息
    process = PythonOperator(
        task_id='process_budget_change',
        op_kwargs={'event': '{{ dag_run.conf }}'},
        ...)
 
# 外部Webhook收到广告主预算变更时调Airflow API触发DAG 2

Airflow的dataset触发（2.4+）

Airflow 2.4引入了Dataset概念——DAG可以声明自己产出或消费的数据集。当产出Dataset的DAG完成后自动触发消费该Dataset的DAG。

from airflow.datasets import Dataset
 
# 定义Dataset
events_dataset = Dataset('s3://data/dwd_events/')
 
# 产出方DAG：完成后标记Dataset更新
with DAG('produce_events', schedule='@daily') as dag_produce:
    task = PythonOperator(
        task_id='write_events',
        outlets=[events_dataset],  # 声明产出这个Dataset
        ...)
 
# 消费方DAG：Dataset更新后自动触发
with DAG('consume_events',
         schedule=[events_dataset]) as dag_consume:  # 依赖Dataset
    task = PythonOperator(task_id='read_events', ...)
 
# 比ExternalTaskSensor更优雅——不需要知道上游DAG的ID
# 只需要知道数据在哪里（Dataset URI）

面试高频追问

追问一：事件驱动和Sensor有什么区别？

Sensor是'拉'模式——下游主动轮询检查上游是否就绪。事件驱动是'推'模式——上游完成后主动通知下游。推模式延迟更低（不用等poke_interval），资源消耗更少（不用Sensor占Worker）。但推模式实现更复杂——需要消息队列或Webhook。Airflow的Dataset是介于两者之间的方案。

追问二：schedule=None的DAG怎么触发？

三种方式：① Airflow UI上手动点Trigger DAG按钮；② REST API调用POST /dags/{dag_id}/dagRuns；③ 另一个DAG通过TriggerDagRunOperator触发。schedule=None意味着这个DAG不会自动调度，只能被外部触发。适合事件驱动和手动任务。

追问三：怎么防止事件触发导致的并发冲突？

如果短时间内触发了多次同一个DAG：① max_active_runs=1保证同时只有一个Run（后续排队）；② 任务幂等保证重复执行不出错；③ 如果业务需要合并多次触发（batch），可以用Airflow的trigger规则或外部队列做消息聚合后再触发。

总结

触发方式讲四种+混合设计：① 定时Cron（常规ETL基础）；② 事件驱动（数据到达/外部变化的即时响应）；③ 手动触发（补数/一次性任务）；④ API触发（系统集成）。生产环境混合使用——Airflow 2.4+的Dataset是优雅的'数据驱动'触发方案。能讲出广告平台三种触发方式混合的架构设计，面试官觉得你的调度设计能适应复杂业务需求。

下一篇聊DAG版本管理：生产环境的DAG怎么用Git管理和CI/CD发布？

你们的任务用什么触发方式？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

调度任务管理 面试通关手册

第16-20篇：工程化管理与面试总结

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长