Apache Airflow 是一个开源的分布式工作流调度与编排平台，用于以编程方式定义、调度和监控复杂的工作流

Apache Airflow 是一个开源的分布式工作流调度与编排平台用于以编程方式定义、调度和监控复杂的工作流DAGsDirected Acyclic Graphs。它使用 Python 编写 DAG支持任务依赖管理、重试机制、并行执行、资源隔离通过 Executor、可扩展的插件生态并能对接多种数据源与计算后端如 Spark、Kubernetes、Celery、Dask 等。核心组件包括DAG有向无环图定义任务及其依赖关系Operator操作符封装具体执行逻辑如PythonOperator、BashOperator、PostgresOperatorExecutor决定任务如何执行如SequentialExecutor本地调试、LocalExecutor、CeleryExecutor分布式、KubernetesExecutor容器级隔离Scheduler解析 DAG、触发任务、维护任务状态Webserver提供 UI 界面用于查看 DAG 运行状态、日志、触发手动运行等Metadata Database存储 DAG 定义、任务实例状态、历史记录等通常用 PostgreSQL 或 MySQLWorkers配合 Celery/K8s实际执行任务的节点。Airflow 不是实时流处理引擎区别于 Flink/Spark Streaming而是面向批处理、ETL、ML pipeline 等有明确起点与终点的定时/事件驱动型任务编排系统。其优势在于可观测性强、社区活跃、插件丰富挑战在于学习曲线较陡、高并发下 Scheduler 可能成为瓶颈需调优或升级至 Airflow 2.0 的改进架构如引入可水平扩展的Standalone Scheduler模式或使用Astronomer等托管方案。# 示例简单 DAG 定义Airflow 2.xfromairflowimportDAGfromairflow.operators.pythonimportPythonOperatorfromdatetimeimportdatetime,timedeltadefprint_hello():print(Hello from Airflow!)withDAG(hello_world,default_args{retries:1},schedule_intervaldaily,start_datedatetime(2024,1,1),catchupFalse,)asdag:taskPythonOperator(task_idprint_hello,python_callableprint_hello,)Airflow Scheduler 性能瓶颈是生产环境中常见问题尤其在 DAG 数量多100、任务并发高、元数据频繁更新或依赖复杂时。以下是常见原因及对应优化策略✅常见瓶颈原因DAG 解析过载Scheduler 频繁扫描dags_folder中大量.py文件解析耗 CPU/IO尤其当 DAG 文件含冗余逻辑、未使用dag装饰器惰性定义、或存在import时执行副作用如网络请求。数据库压力大高频查询/更新task_instance、dag_run、serialized_dag表锁竞争如 PostgreSQL 的行锁、慢查询、缺少索引或连接池不足。Scheduler 单点串行瓶颈Airflow 2.4默认单进程调度器需串行处理 DAG 处理、任务状态同步、触发器检查等2.4 引入Standalone Scheduler模式支持多进程调度器--num-scheduler-pods但需配置得当。过多活跃 DAG Run / Task Instancemax_active_runs_per_dag或全局max_active_tasks设置过高导致状态检查与决策开销剧增。外部依赖阻塞Scheduler 在执行ExternalTaskSensor或调用TriggerDagRunOperator时等待远程 DAG 状态造成线程阻塞尤其未设 timeout 或 poke_interval 不合理。Serialized DAG 同步延迟若启用store_serialized_dagsTrue推荐但min_serialized_dag_update_interval过小会引发频繁 DB 写入。✅关键优化策略DAG 层面使用dag(schedule..., catchupFalse)显式定义避免动态生成 DAG 导致重复解析启用dag_discovery_safe_modeTrueAirflow 2.6跳过语法错误 DAG防止解析中断将非 DAG 逻辑如工具函数、配置加载移出 DAG 文件或用if __name__ __main__:包裹合理设置dagrun_timeout和schedule_interval避免堆积历史 DagRun。Scheduler 配置升级至 Airflow ≥2.4启用scheduler_health_check_thresholdstandalone_scheduler模式配合--num-scheduler-podsN实现横向扩展调整parsing_processes默认2、min_file_process_interval默认0秒 → 建议30–120s、dag_dir_list_interval默认300s降低解析频率启用use_row_level_lockingTruePostgreSQL并确保serialized_dag表有合适索引如idx_serialized_dag_dag_id。数据库优化为高频查询字段添加索引如task_instance.state,dag_run.state,dag_run.execution_date使用连接池如pgbouncer缓解连接数压力定期清理历史数据airflow db clean --clean-before-timestamp。其他实践将 Sensor 类任务设为modereschedule避免占用 worker slot使用TriggerDagRunOperator替代轮询式跨 DAG 依赖监控指标scheduler_heartbeat、dagbag_import_errors、scheduler_num_task_instances、DBslow_queries。 提示Airflow 2.8 引入AsyncExecutor实验性和更细粒度的SchedulerJob分片能力可进一步解耦调度逻辑。