Apache Airflow 这个项目但凡做过数据管道调度的人大概率都绕不开它。4.6 万 Star 放在那儿社区活跃度、生态完整度、招聘市场认可度都是实打实的。但我想聊的不是它有多好——官方文档已经写得够清楚了。我想聊的是当你真正把 Airflow 放进生产环境从一台开发机上的airflow standalone走到几十个 DAG、上百个 Task 每天跑几千次调度的时候哪些设计会救你哪些坑会埋你。这篇文章面向的是已经决定用 Airflow、或者正在评估是否要引入它的工程师。我会从架构层面拆解它的调度机制、Executor 选型逻辑、DAG 解析的隐藏成本然后重点讲落地过程中最容易翻车的几个地方——那些官方文档不会用大标题警告你、但踩过一次就再也不想踩第二次的坑。全文基于 Python 生态涉及大量实操配置和参数取舍建议边看边对照自己的环境。1. 从 Scheduler 的心跳说起Airflow 调度到底是怎么转起来的很多人用 Airflow 用了半年对它的调度机制仍然只有一个模糊印象到点了就触发。但当你遇到 DAG 延迟、任务堆积、Scheduler 假死这些问题时如果不理解调度器内部的心跳循环排查基本靠猜。所以这一节先把调度核心讲透。1.1 Scheduler 的循环逻辑与 DAG 解析开销Airflow 的 Scheduler 本质上是一个无限循环进程每一轮循环做几件事扫描 DAG 目录、解析 DAG 文件、检查调度时间、创建 DagRun、把 TaskInstance 推给 Executor。这个循环的间隔由scheduler_heartbeat_sec控制默认值是 5 秒Airflow 2.x 中实际调度循环由[scheduler] scheduler_heartbeat_sec和min_file_process_interval共同影响。关键在于每一轮循环Scheduler 都要重新解析 DAG 文件。这不是读缓存而是真正执行你的 Python 文件把 DAG 对象构建出来。这意味着如果你的 DAG 文件里写了耗时的顶层代码——比如在模块级别调用了一个 API、查了一次数据库、导入了一个重型库——那么每一轮解析都会付出这个代价。我见过最典型的一个案例某团队的 DAG 文件顶部写了from my_company.ml_pipeline import *而这个模块在导入时会加载一个 200MB 的模型文件。结果就是 Scheduler 每 30 秒卡死一次整个调度延迟从秒级退化到分钟级。排查了半天才定位到是 import 的锅。正确的做法是DAG 文件的顶层代码只做 DAG 定义所有耗时操作放进 Operator 的execute()方法或 PythonOperator 的 callable 里。数据库连接、API 调用、文件读取统统延迟到任务运行时。# 错误示范顶层导入重型模块 from heavy_ml_lib import load_model # 导入即加载 200MB 模型 model load_model() # 顶层执行每次解析都跑 # 正确示范延迟到任务执行时 def run_inference(**context): from heavy_ml_lib import load_model model load_model() # ... 推理逻辑1.2 DAG 文件解析频率的调优参数Airflow 提供了几个参数来控制解析行为理解它们之间的关系很重要参数默认值作用调优建议min_file_process_interval30 秒同一文件两次解析的最小间隔DAG 多时调大到 60-120 秒dag_dir_list_interval300 秒扫描 DAG 目录发现新文件保持默认或调大parsing_processesCPU 核数并行解析进程数按 CPU 核数设置不要超scheduler_heartbeat_sec5 秒调度循环心跳一般不动除非调度压力极大这里有个反直觉的点min_file_process_interval调大并不会让你的任务延迟触发。因为已经解析过的 DAG 结构缓存在数据库中Scheduler 判断是否该触发任务靠的是数据库里的 DagRun 和 TaskInstance 状态不需要重新解析文件。解析只是为了发现 DAG 定义的变更。所以把min_file_process_interval从 30 秒调到 120 秒对调度实时性几乎没有影响但能显著降低 Scheduler 的 CPU 占用。1.3 调度延迟的常见来源排查当你发现任务该跑了但没跑按这个顺序排查Scheduler 是否存活airflow jobs check --job-type SchedulerJob看心跳时间DAG 是否被正确解析airflow dags list看目标 DAG 在不在airflow dags report看解析状态DagRun 是否创建查dag_run表看有没有对应时间点的记录TaskInstance 是否入队查task_instance表看状态是不是queuedExecutor 是否消费如果是 CeleryExecutor看 worker 是否在线、队列是否堆积这个链路走一遍90% 的调度问题都能定位。我特别想强调的是第 2 步——DAG 解析失败是静默的。如果 DAG 文件有语法错误或导入异常Scheduler 不会报错退出只是这个 DAG 不会出现在dags list里。很多人第一次遇到这个问题时以为是调度器坏了其实只是自己的 DAG 文件写挂了。2. Executor 选型从 Sequential 到 Kubernetes 的决策树Executor 是 Airflow 落地时第一个必须做的架构决策而且这个决策一旦定了后面迁移成本很高。我见过太多团队一开始用 LocalExecutor 凑合任务量上来后被迫迁移到 CeleryExecutor结果发现要重新搞一套 Redis/RabbitMQ 集群运维复杂度陡增。2.1 四种主流 Executor 的真实适用边界先把结论摆出来再说理由Executor并行能力运维复杂度适用场景SequentialExecutor单任务串行极低本地开发、调试LocalExecutor单机多进程低单机部署、中小规模CeleryExecutor多机分布式中高大规模、需要弹性伸缩KubernetesExecutor每任务一 Pod高已有 K8s 集群、任务资源隔离要求高SequentialExecutor 只适合开发环境它连 SQLite 都能跑但生产环境绝对不能用。原因很简单它一次只能跑一个任务一个卡住的任务会阻塞整个调度。LocalExecutor 是被低估的选项。很多人觉得它不够生产级但实际上如果你的任务量在每天几千次以内单机 8 核 16G 的配置完全扛得住。它的原理是 Scheduler 进程 fork 出子进程来执行任务没有额外的消息队列组件。我有个项目用 LocalExecutor 跑了两年每天 3000 任务从来没出过并行度问题。它的真正瓶颈是单机资源上限——当你的任务需要不同的 Python 依赖、或者单个任务吃满内存时LocalExecutor 就撑不住了。CeleryExecutor 的核心价值是横向扩展。它把任务通过消息队列分发给多个 WorkerWorker 可以动态增减。但代价是你需要维护 Redis 或 RabbitMQ还要处理 Worker 掉线、队列堆积、任务重复消费这些问题。选它之前先问自己我的任务量真的需要多机吗如果单机扛得住别为了看起来更专业而上 Celery。KubernetesExecutor 适合任务资源需求差异极大的场景。每个 TaskInstance 起一个独立的 Pod任务之间完全隔离资源按需分配。缺点是 Pod 启动有延迟通常 10-30 秒对于大量短任务来说这个开销可能比任务本身还长。所以它更适合任务少但每个任务重的场景比如跑一个需要 16G 内存的 Spark 作业。2.2 CeleryExecutor 的队列设计经验如果你确定要用 CeleryExecutor队列划分是个必须提前想清楚的事。默认情况下所有任务都进default队列这会导致一个重型任务把 Worker 占满后面的轻量任务全部排队。我的做法是按资源特征而不是按业务线划分队列# 按资源特征划分队列 heavy_task PythonOperator( task_idtrain_model, queueheavy, # 高内存队列Worker 配置 32G ... ) light_task PythonOperator( task_idsend_notification, queuelight, # 轻量队列Worker 配置 2G ... )然后给不同队列配置不同规格的 Worker。这样重型任务不会挤占轻量任务的资源整体吞吐量能提升不少。踩过的坑是队列名一旦上线就不要改因为历史 TaskInstance 里记录的是旧队列名改了之后这些任务会找不到 Worker。2.3 KubernetesExecutor 的 Pod 模板复用技巧KubernetesExecutor 最烦人的地方是每个任务都要拉镜像、起 Pod冷启动慢。优化手段是配置pod_template_file把公共的镜像、环境变量、资源限制抽出来# pod_template.yaml apiVersion: v1 kind: Pod metadata: name: airflow-worker spec: containers: - name: base image: my-registry/airflow-worker:2.7.0 env: - name: AIRFLOW__CORE__EXECUTOR value: KubernetesExecutor resources: requests: memory: 512Mi cpu: 250m然后在 DAG 里通过executor_config覆盖单个任务的资源需求。这样基础镜像可以预装常用依赖减少每次拉取的时间。另外如果 K8s 集群支持镜像缓存把 Worker 镜像预热到各个节点上冷启动能从 30 秒降到 10 秒以内。3. DAG 设计里那些看起来没问题的写法DAG 写得好不好短期看不出来跑上三个月问题全暴露。这一节讲几个我踩过的设计坑都是那种当时觉得挺合理后来发现是灾难的写法。3.1 动态 DAG 生成的边界用循环批量生成 DAG 是个很自然的想法比如给 50 个客户各生成一个 ETL DAG# 这种写法在 DAG 数量少时没问题 for client in clients: dag DAG(fetl_{client}, ...) # 定义任务 globals()[fetl_{client}] dag问题在于每次解析 DAG 文件这个循环都要跑一遍。如果clients是从数据库查出来的那每次解析都要查一次库。50 个客户还好500 个客户时 Scheduler 就吃不消了。更麻烦的是这种写法生成的 DAG 在 Web UI 里是一堆独立的 DAG管理起来很痛苦。我的建议是能用单个 DAG 动态任务就用单个 DAG。比如用TaskGroup或者动态生成 Taskwith DAG(etl_all_clients, ...) as dag: for client in clients: task PythonOperator( task_idfetl_{client}, python_callablerun_etl, op_kwargs{client: client}, )这样只有一个 DAG但内部有多个任务管理清晰解析开销也小。如果客户数量是动态的可以用expand()做动态任务映射Airflow 2.3 支持task def process_client(client): # 处理逻辑 pass clients_list [client_a, client_b, ...] process_client.expand(clientclients_list)3.2 任务幂等性重跑是常态不是异常Airflow 的任务重跑太常见了——手动 clear、失败重试、补数据都会导致同一个任务跑多次。如果你的任务不是幂等的重跑就会产生脏数据。我见过最惨的一个案例一个任务负责给用户账户加 100 积分重跑一次就多加 100。补数据时跑了 5 次用户凭空多了 500 积分最后只能人工回滚。幂等性的实现方式取决于任务类型写数据库用INSERT ... ON CONFLICT DO UPDATE或先DELETE再INSERT保证同一批次数据只写一次写文件用临时文件 原子重命名或者按执行日期分区覆盖调 API用业务唯一键做去重或者让 API 支持幂等 tokenAirflow 提供了execution_date2.x 中叫logical_date作为天然的去重键善用它def write_data(**context): logical_date context[logical_date] # 先删除该日期分区 db.execute(DELETE FROM metrics WHERE dt %s, (logical_date.date(),)) # 再写入 db.execute(INSERT INTO metrics ...)3.3 用 XCom 传大数据的代价XCom 是 Airflow 任务间传数据的机制但它的设计初衷是传小数据——比如一个文件路径、一个状态标记。如果你用它传 DataFrame 或者大 JSON会出问题。XCom 的数据存在元数据库里默认是 PostgreSQL 或 MySQL传大数据会导致元数据库体积膨胀查询变慢Scheduler 解析 XCom 时内存占用飙升数据库连接被长时间占用我见过有人用 XCom 传一个 50MB 的 DataFrame结果元数据库一周涨了 10GScheduler 响应明显变慢。正确的做法是大数据落地到对象存储或共享文件系统XCom 只传路径。def extract(**context): df fetch_data() path fs3://bucket/data/{context[logical_date]}.parquet df.to_parquet(path) return path # XCom 只传路径 def transform(**context): path context[ti].xcom_pull(task_idsextract) df pd.read_parquet(path) # 处理逻辑如果确实需要传中等大小的数据可以配置 XCom 后端为 S3 或 GCSAirflow 2.x 支持自定义 XCom Backend把数据存到对象存储元数据库只存引用。4. 元数据库Airflow 最容易被忽视的性能瓶颈Airflow 的所有状态——DAG 定义、DagRun、TaskInstance、XCom、连接信息——都存在元数据库里。这个数据库的性能直接决定了整个 Airflow 的响应速度。但很多人在部署时随便给个 MySQL 就完事了跑一段时间后 Web UI 卡顿、Scheduler 延迟才发现是数据库的问题。4.1 元数据库的选型与配置底线SQLite 只能用于开发这个没有商量余地。它的并发写入能力极差Scheduler 和 Web Server 同时访问就会锁表。生产环境用 PostgreSQL 或 MySQL 都行但有几个配置必须调# PostgreSQL 关键配置 # postgresql.conf max_connections 200 # 默认 100 不够用 shared_buffers 2GB # 建议为内存的 25% work_mem 16MB # 排序和哈希操作的内存Airflow 侧的连接池配置# airflow.cfg [sql_alchemy] sql_alchemy_pool_size 10 sql_alchemy_max_overflow 20 sql_alchemy_pool_recycle 1800 # 30 分钟回收连接sql_alchemy_pool_recycle这个参数特别重要。如果数据库或中间有负载均衡器会断开空闲连接不设置回收时间的话Airflow 会拿到失效连接然后报错。我遇到过好几次随机报数据库连接错误最后都是这个参数没配导致的。4.2 元数据库的清理策略Airflow 不会自动清理历史数据dag_run、task_instance、xcom、log这些表会无限增长。一个中等规模的 Airflow 实例跑一年后元数据库几十 G 是常事。清理方式有两种方式一用airflow db clean命令# 清理 90 天前的数据 airflow db clean --clean-before-timestamp 2024-01-01 --tables task_instance,dag_run,xcom方式二配置自动清理Airflow 2.6# airflow.cfg [scheduler] # 自动清理超过 90 天的元数据 clean_tis_without_dagrun_interval 90我的经验是保留 30-90 天的元数据就够了更早的数据导出到数据仓库做审计。清理时注意顺序先清task_instance和xcom再清dag_run否则会有外键约束问题。另外log表存的是任务日志的元信息不是日志内容本身日志内容默认存在本地文件系统。如果日志量大建议配置远程日志存储S3、GCS、OSS否则本地磁盘很快会被写满。4.3 连接池耗尽与长事务问题Airflow 的 Scheduler、Web Server、Worker 都会连元数据库。当并发任务多时连接池很容易耗尽表现为QueuePool limit of size X overflow Y reached。除了调大连接池更根本的解决方式是减少长事务。Airflow 有些操作会持有数据库连接较长时间比如大批量 TaskInstance 状态更新XCom 的读写DAG 解析时的批量写入如果发现连接池频繁耗尽可以查一下数据库的慢查询日志看看是不是有长事务。另外[core] parallelism和[core] max_active_tasks_per_dag这两个参数控制并发度调太大也会加剧连接竞争。5. 生产环境部署的运维细节前面讲的都是设计层面的事这一节讲部署和运维中那些具体的、琐碎的、但会直接影响稳定性的细节。5.1 时间同步与时区陷阱Airflow 的调度严重依赖时间。如果服务器时间不同步会出现任务提前触发、延迟触发、甚至重复触发的问题。所有 Airflow 节点必须配置 NTP 时间同步这是底线。另外Airflow 内部统一用 UTC 时间但 DAG 的start_date和schedule_interval可以指定时区。这里有个经典坑# 危险写法用 naive datetime from datetime import datetime dag DAG(my_dag, start_datedatetime(2024, 1, 1), schedule_intervaldaily) # 安全写法用带时区的 datetime import pendulum dag DAG(my_dag, start_datependulum.datetime(2024, 1, 1, tzAsia/Shanghai), schedule_intervaldaily)用 naive datetime 时Airflow 会按 UTC 解释导致你以为是北京时间 0 点跑实际是 UTC 0 点北京时间 8 点跑。这个坑我踩过排查了半天才发现是时区问题。5.2 日志管理与磁盘水位Airflow 的任务日志默认存在$AIRFLOW_HOME/logs下按 DAG ID 和 Task ID 分目录。如果不做清理磁盘很快会被写满。三个层面的处理配置远程日志把日志写到 S3/GCS/OSS本地只保留最近几天的配置日志轮转用 logrotate 或 Airflow 自带的日志清理监控磁盘水位磁盘使用率超过 80% 就告警远程日志配置示例# airflow.cfg [logging] remote_logging True remote_log_conn_id my_s3_conn remote_base_log_folder s3://my-bucket/airflow-logs配置远程日志后Web UI 查看日志时会从远程拉取本地磁盘压力大大减轻。但要注意远程日志的读取权限要配好否则 Web UI 会报权限错误。5.3 高可用部署的取舍Airflow 的高可用主要涉及三个组件Web Server可以起多个实例前面挂负载均衡无状态好做SchedulerAirflow 2.x 支持多 Scheduler 实例HA 模式但需要额外的配置和数据库锁机制WorkerCeleryExecutor 天然支持多 WorkerKubernetesExecutor 靠 K8s 调度多 Scheduler 的配置# airflow.cfg [scheduler] # 启用 HA 模式 use_row_level_locking True启用后可以起多个 Scheduler 进程它们通过数据库行锁协调避免重复调度。但要注意多 Scheduler 对元数据库的压力更大数据库性能不够时反而会降低稳定性。我的建议是中小规模用单 Scheduler 监控告警大规模再上多 Scheduler。6. 那些官方文档不会重点讲的踩坑实录这一节是我个人和团队在实际项目中踩过的坑按问题现象 → 排查过程 → 根因 → 解决方案的结构呈现希望能帮你少走弯路。6.1 任务卡在 queued 状态一个被忽视的 Worker 配置现象CeleryExecutor 环境下任务创建后一直卡在queuedWorker 日志没有任何输出。排查过程先看 Worker 是否在线airflow celery status显示在线再看队列是否有堆积airflow celery inspect active显示队列为空最后看任务定义的 queue 参数发现任务指定了queuegpu但没有任何 Worker 监听gpu队列。根因Worker 启动时通过-q参数指定监听的队列默认只监听default。任务指定了不存在的队列就永远没人消费。解决方案要么给 Worker 加上对应队列的监听要么把任务的 queue 参数改回default。建议在 CI 里加一个检查确保所有任务引用的队列都有对应的 Worker。6.2 DAG 突然消失一个 import 引发的血案现象某个 DAG 在 Web UI 里突然不见了但文件明明还在。排查过程airflow dags list确认 DAG 不在列表里查看 Scheduler 日志发现Broken DAG错误错误信息指向 DAG 文件里的一个 import 语句。根因DAG 文件里 import 了一个第三方库这个库在某个 Worker 节点上没装。Scheduler 解析 DAG 时执行 import 失败整个 DAG 被标记为 broken不会出现在列表里。解决方案确保所有 Airflow 节点Scheduler、Worker、Web Server的 Python 环境一致。用 Docker 镜像统一环境是最稳妥的做法。另外DAG 文件里的 import 尽量放在函数内部减少解析时的依赖。6.3 补数据把数据库打挂了并发控制的教训现象为了补一个月的数据手动触发了 30 个 DagRun结果元数据库 CPU 飙到 100%整个 Airflow 无响应。排查过程查数据库慢查询日志发现大量task_instance的插入和更新操作查 Airflow 并发配置发现max_active_tasks_per_dag设的是 1630 个 DagRun 同时跑就是 480 个任务并发。根因补数据时没有限制并发瞬间产生大量任务元数据库扛不住。解决方案补数据时用max_active_runs限制同时运行的 DagRun 数量dag DAG( my_dag, max_active_runs3, # 同时最多 3 个 DagRun ... )另外可以用 Airflow 的backfill命令它支持--max-active-runs参数控制并发。补数据是个资源密集型操作最好安排在业务低峰期并且提前和数据库团队打招呼。6.4 时区问题导致的重复调度现象一个daily的 DAG在某个时间点触发了两次。排查过程查dag_run表发现同一logical_date有两条记录查 Scheduler 日志发现两个 Scheduler 实例都在调度这个 DAG。根因部署了两个 Scheduler 实例但没有启用 HA 模式的行级锁导致两个实例同时判断该触发了各创建了一个 DagRun。解决方案启用use_row_level_locking True或者只部署一个 Scheduler。多 Scheduler 虽然能提高可用性但配置不当反而会引入重复调度问题。7. 写在最后Airflow 的边界在哪里用了几年 Airflow我越来越清楚它适合什么、不适合什么。它适合有明确调度周期的批处理任务、需要依赖管理的 ETL 管道、需要可视化监控和重跑能力的数据工作流。它的 DAG 抽象、丰富的 Operator 生态、成熟的 Web UI在批处理调度这个领域确实很难找到替代品。它不适合毫秒级延迟的实时任务用消息队列或流处理框架、超大规模的任务编排几万个任务同时跑元数据库会成为瓶颈、需要复杂条件分支和动态拓扑的场景DAG 是静态定义的动态性有限。我个人的经验是Airflow 的复杂度主要不在写 DAG而在运维。一个跑得稳的 Airflow 集群背后是合理的 Executor 选型、调优过的元数据库、规范的 DAG 编写约定、完善的监控告警。如果你只是想让几个脚本按时跑起来可能 cron 一个简单的日志系统就够了。但如果你需要管理几十上百个有依赖关系的任务需要重跑、补数据、可视化那 Airflow 值得你投入时间去理解它的内部机制。最后分享一个我一直在用的检查清单每次上线新 DAG 前过一遍DAG 文件顶层有没有耗时操作任务是否幂等重跑会不会产生脏数据XCom 传的是不是小数据有没有设置max_active_runs和retries时区用的是不是带时区的 datetime任务引用的队列有没有对应的 Worker元数据库的清理策略配了吗这七个问题看起来简单但每一个背后都是真金白银的教训。
