第一次打开 Airflow 的 Web UI绝大多数人的注意力都会停在 Graph View 那张五颜六色的图上。绿色代表任务成功红色代表失败蓝色代表运行中灰色代表排队等待还有被跳过、被上游阻断等特殊状态。很多人第一眼会觉得这界面有点“花里胡哨”而我的判断恰恰相反Airflow 之所以能从 Airbnb 内部工具成长为 Apache 顶级项目从“批量脚本调度器”变成数据平台事实标准的编排引擎最大功臣不是 DAG 语法也不是庞大的 Operator 生态而是这张“带颜色的图”背后的一整套任务状态机与调度语义。从工程本质看cron 和 Airflow 解决的问题完全不同。cron 只保证“到点执行一条命令”它不关心任务是否成功不知道任务之间是否有依赖也不能把失败批次挑出来重跑。Airflow 把数据管道从一张时间表变成一张有向无环图每个节点可以被观察、被重试、被并行调度、被限流而这些能力最终都汇聚在 UI 上那一个个颜色块里。所以题目中说 Airflow 是 built with Colors并不是在夸界面好看而是在说它把调度系统最核心的可观测性做成了默认能力。这篇文章会从痛点切入讲清楚 Airflow 的核心架构、核心概念、单机安装部署、第一个 DAG 的完整实现、生产环境改造思路、常见问题排查和最佳实践。文章示例基于 Airflow 2.x 编写3.x 的 DAG 定义方式整体保持一致具体安装版本请以官方发布为准。1. 这篇文章真正要解决的问题先看一个很常见的场景。某个数据团队维护着 30 多个定时任务任务是典型的“链路式”结构每天凌晨从业务库抽取数据经过清洗后写入数仓再触发指标计算最后给报表平台推送结果。最早的时候团队用 cron 写了一批脚本每个脚本负责一个环节。开始只有几个任务大家都觉得挺方便。随着任务变多问题开始集中爆发A 环节凌晨 2 点跑挂了B 环节 3 点照常启动读到的是不完整数据而且没有自动补偿。某天上游数据延迟到 4 点才到位下游脚本仍然严格按 cron 时间执行结果整条链路白跑。任务失败后运维要登录服务器手工重跑还要记住重跑命令时间一长就乱。想并行处理 6 个互不依赖的数据源cron 只能按固定时间串起来或者靠后台进程自己管理并发。老板问“今天的数据为什么没出”你只能看服务器日志没人能一眼说清失败发生在哪个环节。这些问题可以归纳为三个层次。第一层是依赖编排。任务之间不是孤立的A 成功之后才能跑 BB 失败后 C 不应该启动。cron 解决不了跨任务依赖只能靠脚本内部自己判断判断逻辑一旦分散就容易出现脏数据。第二层是状态管理。每个任务实例都应该有明确状态排队、运行中、成功、失败、跳过、被上游阻断。有了状态系统才能决定下一步动作才能支持失败重试、定时重跑、补历史数据。第三层是可观测性。调度系统不能只“执行”还必须让操作者随时看到当前工作流进行到哪一步。Graph View 里每个 Task 的颜色本质上就是把状态机可视化让“看日志猜原因”变成“看颜色定位问题”。Airflow 恰好在这三个层次上都提供了标准答案。如果你需要的是一个能处理跨任务依赖、支持失败重跑、能在分布式环境中扩展、并且有完整 Web UI 的开源调度平台Apache Airflow 是目前最值得优先考虑的选择。2. 核心概念DAG、Task、Operator 与 Executor在动手安装之前先把 Airflow 的核心概念讲清楚。很多初学者卡住不是卡在代码语法上而是没搞清楚对象之间的关系。2.1 DAG 是静态拓扑不是执行队列DAGDirected Acyclic Graph有向无环图是用 Python 代码描述的一整套工作流依赖关系。比如“先抽数、再清洗、再计算”就是三个节点和两条有向边。关键点在于DAG 本身只是一种拓扑描述它不会“执行任务”它只负责告诉 Scheduler“谁在谁前面”。这也解释了为什么 Airflow 的 DAG 文件是 Python 文件。Python 代码天然适合表达依赖关系而且可以在循环里动态生成 Task后面示例会展示这一点。2.2 Operator 是任务模板Task 是具体实例Operator 是 Airflow 预置或自定义的任务执行模板。BashOperator 用来执行 shell 命令PythonOperator 用来调用 Python 函数PostgresOperator 用来执行 SQLDockerOperator 用来运行 Docker 容器。你可以把 Operator 理解成“工具类”。同一个 Operator 在 DAG 中出现一次就生成一个 Task。比如一个 PythonOperator 模板在 DAG 里写了三个 Task它们功能相似但属于三个独立执行单元。再往下还有 TaskInstance 的概念它代表“某个 Task 在某个时间点的某次运行”。一个 Task 可以运行很多次每一次都是一个 TaskInstance每个 TaskInstance 都有独立状态。Graph View 里那个颜色块实际指示的是 TaskInstance 的状态。2.3 Executor 是真正干活的组件Executor 决定了 TaskInstance 用什么方式执行。这是 Airflow 架构里非常关键的一环。SequentialExecutor默认执行器同一时间只能跑一个任务适合本地调试生产环境不要用。LocalExecutor单机多进程执行并行度较高适合小规模部署。CeleryExecutor通过 Celery 分布式队列把任务分发给多个 Worker是生产环境最常见的方案。KubernetesExecutor每个 TaskInstance 动态创建一个 Kubernetes Pod 执行资源隔离好适合容器化平台。Scheduler 负责解析 DAG、判断哪些 Task 可以运行、把任务交给 ExecutorExecutor 负责真正调度执行。理解这个关系后面排查“任务一直 queued 不运行”时会非常有用。2.4 其他常用对象Sensor一种特殊的 Operator用来等待某个条件满足。比如等待文件生成、等待外部接口就绪。Hook封装外部系统连接的对象是 Operator 与外部系统交互的通道。XComTask 之间传递小消息的机制适合传递状态、路径、参数不适合传大数据。VariableAirflow 全局变量用于配置管理。Connection统一存储外部系统的连接信息包括主机、账号、密码避免在 DAG 代码里硬编码敏感信息。Triggerer从 Airflow 2.2 开始引入配合 Deferrable Operator 实现轻量级异步等待能显著降低空占资源的 Sensor 对 Scheduler 和 Worker 的压力。2.5 常见概念对比概念一句话理解容易混淆的点DAG工作流依赖图不是任务执行队列TaskOperator 在 DAG 中的实例不是具体某一趟运行TaskInstanceTask 在某一次调度中的运行同一 Task 可以有多个实例Operator可复用的执行模板不是运行实体Executor决定任务如何执行不是 Worker 本身Scheduler判断任务何时执行不真正执行业务逻辑XComTask 间传消息不适合传 DataFrame3. 环境准备与前置条件单机快速体验 Airflow比大多数人想象中简单。你只需要一个 Python 环境建议使用 Python 3.8 及以上版本。生产环境推荐 Linux PostgreSQL Redis但第一步只需要跑通本地。3.1 创建独立虚拟环境Airflow 依赖比较多强烈建议使用虚拟环境避免污染系统 Python。sudo apt update sudo apt install -y python3-venv python3-dev build-essential sudo mkdir -p /opt/airflow sudo chown -R $USER:$USER /opt/airflow python3 -m venv /opt/airflow/venv source /opt/airflow/venv/bin/activate注意如果你的服务器无法访问系统软件源可以只创建 Python 虚拟环境不安装系统依赖包。对大多数纯 Python 使用场景缺少编译工具也不影响 pip 安装。3.2 安装 Apache Airflow安装前设置 AIRFLOW_HOME这个目录会保存 airflow.cfg、日志和 SQLite 数据库文件。export AIRFLOW_HOME/opt/airflow pip install --upgrade pip pip install apache-airflow[celery,redis,postgres]这里安装了带 Celery、Redis 和 PostgreSQL 扩展支持的版本方便后面直接做生产化改造。如果只想快速体验只装apache-airflow也可以。安装完成后用命令确认版本airflow version3.3 初始化元数据库Airflow 默认使用 SQLite 作为元数据库。第一次使用要执行airflow db init这个命令会创建 Airflow 运行所需的表并生成airflow.cfg和webserver_config.py。执行完成后可以看到$AIRFLOW_HOME目录下出现 airflow.db 文件。3.4 创建管理员用户Airflow 2.x 默认启用 RBAC 认证必须创建用户才能登录 Web UI。airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com \ --password admin123生产环境一定要换成强密码并通过环境变量或外部密钥管理方式配置不要直接写在命令历史里。3.5 启动 Web Server 和 Scheduler新开两个终端分别启动两个核心进程。终端一airflow webserver -p 8080终端二airflow scheduler然后浏览器访问http://localhost:8080用刚才创建的账号登录。看到首页为空是很正常的因为还没有任何 DAG。接下来我们写第一个 DAG。4. 第一个 DAG彩虹批处理管道为了让示例贴近实际我用“颜色”作为业务维度构建一个批处理管道先准备数据再按颜色并行处理多个分片最后汇总并结束。这个 DAG 同时演示了动态生成 Task、参数传递、Jinja 模板和任务依赖。4.1 创建 DAG 文件在$AIRFLOW_HOME/dags目录下创建rainbow_pipeline.py。# 文件路径/opt/airflow/dags/rainbow_pipeline.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.empty import EmptyOperator from airflow.operators.python import PythonOperator COLORS [red, orange, yellow, green, blue, purple] def process_color(color: str, **context): 处理某个颜色对应的批次数据。 生产环境中这里可以替换为读取当日分区文件、调用下游接口、写入目标数据表。 ds context[ds] run_id context[run_id] print(frun_id{run_id}, ds{ds}, color{color}) print(fprocessing {color} batch ...) # 实际业务逻辑尽量保证幂等失败后可以安全重跑 default_args { owner: data_platform, depends_on_past: False, retries: 2, retry_delay: timedelta(minutes5), email_on_failure: False, email_on_retry: False, } with DAG( dag_idrainbow_pipeline, description一个用颜色展示 Airflow 状态语义与动态任务生成的最小 DAG, start_datedatetime(2024, 1, 1), scheduledaily, catchupFalse, max_active_runs3, tags[demo, colors], default_argsdefault_args, ) as dag: start EmptyOperator(task_idstart) prepare_data BashOperator( task_idprepare_data, bash_command( mkdir -p /tmp/airflow-demo/{{ ds }} echo source data ready /tmp/airflow-demo/{{ ds }}/source.txt ), ) color_tasks [] for color in COLORS: task PythonOperator( task_idfprocess_{color}, python_callableprocess_color, op_kwargs{color: color}, ) color_tasks.append(task) generate_summary BashOperator( task_idgenerate_summary, bash_command( echo all colors processed /tmp/airflow-demo/{{ ds }}/summary.txt ), ) end EmptyOperator(task_idend) start prepare_data color_tasks generate_summary end4.2 代码逻辑拆解这段代码有几个关键设计。在with DAG(...) as dag:块内创建的 Operator 会自动绑定到该 DAG所以不需要重复传dagdag。这种写法在 Airflow 2.x 中是官方推荐风格。COLORS列表配合for循环动态创建了 6 个process_red、process_orange、process_yellow、process_green、process_blue、process_purple任务。这在实际项目中非常常见业务维度事先不确定数据源是一个动态列表通过循环生成任务可以大幅减少重复代码。BashOperator 里的{{ ds }}是 Jinja 模板变量Airflow 在执行时会把ds替换成当前执行日期格式是YYYY-MM-DD。使用模板变量的好处是任务具备“运行时语义”同一个 Task 在每天的运行中会自动落在不同日期目录下。default_args中的retries和retry_delay是任务失败后的自动重试策略。对依赖外部系统稳定性的任务来说自动重试 1 到 3 次间隔几分钟能显著降低外部抖动导致的误报。4.3 验证 DAG 文件DAG 文件本质上是 Python 文件所以可以直接用 Python 解释器执行来检查语法python /opt/airflow/dags/rainbow_pipeline.py如果没有报错说明 DAG 可以正常解析。更严谨的做法是等 Scheduler 扫描到文件后用 Airflow 命令确认airflow dags list | grep rainbow airflow tasks list rainbow_pipelinetasks list会输出 DAG 中所有 Task 的名字正常情况应该包含start prepare_data process_red process_orange process_yellow process_green process_blue process_purple generate_summary end5. Web UI 操作与运行验证Airflow 最有价值的体验在 Web UI。登录后Pages 列表中会看到rainbow_pipeline。如果列表为空说明 DAG 文件没有被 Scheduler 扫描到这时先看airflow dags list-import-errors有没有导入错误。5.1 Graph View颜色状态图点击 DAG 名称进入 Graph View。这个页面是 Airflow 最经典的视图每个 Task 显示为一个节点节点之间的箭头表示依赖关系。未运行的任务默认是灰色。触发一次运行后节点颜色会随着生命周期变化蓝色任务正在运行。绿色任务成功。红色任务失败。橙色或黄色任务重试中。紫色任务被跳过。灰色任务排队等待。颜色的变化不是装饰而是状态机的可视化。当管道出现红色节点时你可以立刻看到失败发生在哪一环以及上游和下游受影响的范围。5.2 手动触发一次运行点击页面右上角的“播放”按钮或者使用命令行airflow dags trigger rainbow_pipeline触发后点进 Graph View会看到任务开始排队并陆续变成蓝色最终全部变绿。点击任意绿色节点可以查看该 Task 这一次运行的日志。我们可以在日志中看到process_color函数打印的内容。5.3 Grid View看总体运行情况Airflow 2.x 新增的 Grid View 把 DAG 每次运行和每个 Task 的状态放在一张表格里比 Tree View 更直观。这里可以快速确认每日运行是否成功、失败任务集中在哪里是日常巡检最常用的视图。5.4 触发条件与时间规则新手最容易困惑的是“为什么任务没有在我设定的时间运行”。Airflow 的时间规则有两点必须记住。第一start_date是第一个调度周期的起点不是第一次运行时间。对于scheduledaily第一个 Run 的触发时间通常是start_date 一个周期。比如start_datedatetime(2024, 1, 1)第一次运行会在 2024-01-02 凌晨触发而这一次运行的执行日期仍然是 2024-01-01。第二catchup默认是 True。如果start_date设置得比较早Scheduler 发现历史周期从未运行过会不断补跑历史任务。这在某些场景是优点但在大多数新开发场景中是惊吓。示例中显式设置catchupFalse避免产生一堆历史 Run。如果你发现任务“不在预期时间运行”优先检查三处start_date、schedule、时区配置。Airflow 默认使用 UTC如果你的业务在东八区且没有设置timezone凌晨 1 点的任务实际会在 UTC 凌晨 1 点运行也就是北京时间上午 9 点。6. 生产环境部署从单机到分布式单机跑通只是第一步。真实生产环境要考虑高可用、并发能力、数据安全性、日志管理等通常不会继续使用 SQLite 和 SequentialExecutor。6.1 推荐参考架构生产环境最常见的组合是PostgreSQL 存储元数据替代 SQLite。Redis 作为 Celery Broker负责任务消息传递。CeleryExecutor 负责任务分发。一个或多个 Worker 节点真正执行业务任务。Web Server 和 Scheduler 分离部署。可选部署 Flower用于监控 Celery Worker 状态。这个架构里Scheduler 只负责“决策”不负责“执行”。真正跑任务的是 Worker因此业务任务再重也不太会影响调度器稳定性。6.2 docker-compose 快速搭建如果团队已经使用 Docker用 docker-compose 搭建一套开发环境很快。下面是一个简化但可运行的参考。version: 3.8 x-airflow-common: airflow-common image: apache/airflow:2.10 environment: AIRFLOW__CORE__EXECUTOR: CeleryExecutor AIRFLOW__CORE__LOAD_EXAMPLES: false AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresqlpsycopg2://airflow:airflowpostgres:5432/airflow AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0 AIRFLOW__CELERY__RESULT_BACKEND: dbpostgresql://airflow:airflowpostgres:5432/airflow volumes: - ./dags:/opt/airflow/dags depends_on: - postgres - redis services: postgres: image: postgres:14 environment: POSTGRES_USER: airflow POSTGRES_PASSWORD: airflow POSTGRES_DB: airflow redis: image: redis:7 webserver: : *airflow-common ports: - 8080:8080 command: webserver scheduler: : *airflow-common command: scheduler worker: : *airflow-common command: celery worker说明几点。通过环境变量方式配置 Airflow是比直接改airflow.cfg更推荐的做法尤其是容器化场景。环境变量命名规则是AIRFLOW__SECTION__KEY例如AIRFLOW__CORE__EXECUTOR对应[core] executor。x-airflow-common是 YAML 锚点语法用于复用公共配置。实际部署时镜像 tag 请以官方可用版本为准不要直接拿生产环境使用latest。6.3 systemd 方式部署不依赖 Docker 的环境可以用 systemd 管理 Airflow 进程。以 Linux 用户airflow为例创建两个服务单元文件。/etc/systemd/system/airflow-webserver.service[Unit] DescriptionAirflow web server Afternetwork.target Wantsnetwork.target [Service] Typesimple Userairflow Groupairflow EnvironmentAIRFLOW_HOME/opt/airflow ExecStart/opt/airflow/venv/bin/airflow webserver -p 8080 Restarton-failure RestartSec10s [Install] WantedBymulti-user.target/etc/systemd/system/airflow-scheduler.service[Unit] DescriptionAirflow scheduler Afternetwork.target Wantsnetwork.target [Service] Typesimple Userairflow Groupairflow EnvironmentAIRFLOW_HOME/opt/airflow ExecStart/opt/airflow/venv/bin/airflow scheduler Restarton-failure RestartSec10s [Install] WantedBymulti-user.target启动并设置开机自启sudo systemctl daemon-reload sudo systemctl enable --now airflow-webserver sudo systemctl enable --now airflow-scheduler生产环境还应该配置日志定期清理、Worker 健康检查和元数据库备份。6.4 生产环境关键配置项以下配置项直接影响调度稳定性需要重点确认。配置项作用建议[core] executor指定执行器生产使用 CeleryExecutor 或 KubernetesExecutor[core] dags_folderDAG 文件目录固定路径避免误改[core] load_examples是否加载官方示例 DAG设置为 False[core] timezone全局时区按团队规范设置建议统一[core] parallelism全局最大并行任务数根据 Worker 资源调整[database] sql_alchemy_conn元数据库连接使用 PostgreSQL 并配置连接池[celery] broker_urlCelery 消息队列地址使用 Redis配置密码[celery] result_backendCelery 结果存储使用数据库后端[celery] worker_concurrency单个 Worker 并行任务数根据任务负载调优7. 常见问题与排查思路Airflow 使用中踩坑非常正常很多问题看起来千奇百怪根源就那么几类。整理几个高频问题供读者对照排查。问题现象可能原因排查方式解决方案Web UI 没有看到 DAGDAG 文件路径不对或 Python 语法错误airflow dags list-import-errors修复 DAG 文件确认dags_folder路径任务一直处于 queued不运行Executor 配置不对Worker 未启动或并发数占满查看 Scheduler 日志、Worker 状态确认使用 CeleryExecutor 并启动 Worker检查 parallelism任务运行时间不符合预期时区没设置或 start_date/schedule 理解有偏差查看 Run 的 execution_date统一时区配置按 Airflow 周期规则设计任务失败后自动重跑了一堆历史任务catchup 默认为 True查看 DAG 的 Run 列表新项目设置 catchupFalseDAG 文件刚修改UI 不刷新Scheduler 每分钟扫描一次 DAG 文件等待或手动重启 Scheduler了解 DAG 扫描机制避免频繁改 DAG调用外部服务报连接失败Connection 配置错误或密码被加密在 UI 的 Admin Connections 中测试正确配置 Connection确认 fernet_keyWorker 内存持续上涨任务并发过高或单个任务加载了太多数据用监控工具查看 Worker 资源调整 worker_concurrency优化任务代码部署后账号密码失效web server secret key 或 fernet_key 变化查看 Web UI 日志固定配置密钥备份 secrets7.1 DAG 导入错误是最高频问题Airflow 的 Scheduler 会周期性扫描 DAG 文件目录。只要文件里有未捕获的异常整个文件对应的 DAG 都会被标记为导入失败。排查时第一时间执行airflow dags list-import-errors如果输出里有错误信息按照 Python traceback 处理即可。常见原因包括顶层代码访问了不存在的环境变量、依赖包未安装、文件里写了与 Airflow 无关的启动逻辑。7.2 任务排队不执行任务一直 queued是生产环境最常见的“假死”现象。在 CeleryExecutor 环境下先检查 Worker 进程是否存在ps -ef | grep celery worker再检查[celery] worker_concurrency是否被占满。一个 Worker 的并发数是有限的如果所有并发槽都被长任务占满新任务就会一直排队。此时需要增加 Worker 节点或者调优任务执行时间。7.3 日志和元数据越来越庞大Airflow 运行时间越长日志和元数据表增长越快。建议定期清理历史日志并考虑将日志接入集中式日志平台。元数据表里的 TaskInstance 数据也可以按保留策略清理避免数据库膨胀拖慢 Scheduler 扫描。8. 最佳实践与工程建议要把 Airflow 在团队里用好光会写 DAG 还不够。以下建议来自大量生产实践基本是按优先级排序的。8.1 任务必须幂等这是所有数据管道的第一原则。同一个 Task 无论重跑多少次结果应该一致。典型做法是写入数据前先清理目标分区再写入当日新数据文件任务先写临时目录成功后再移动到最终目录。幂等设计能保证失败重跑、补数、修复数据时不会产生脏数据。8.2 一个 Task 只做一件事把一个大任务拆成多个小任务会增加调度开销但收益更大失败定位更精确重试粒度更小可读性更好。理想情况下一个 Task 失败后单独重跑它就能恢复不需要重跑整条链路。8.3 敏感信息不硬编码数据库密码、接口密钥、云厂商凭证不应该出现在 DAG 文件里也不应该写在default_args中。Airflow 的 Connections 和 Variables 就是为此设计的。生产环境建议配合外部密钥管理服务使用。8.4 用 Pool 控制并发资源如果多个 DAG 共享同一套资源比如同一个数据库集群使用 Pool 可以限制同时运行的任务数。from airflow.utils.state import State from airflow.operators.python import PythonOperator task PythonOperator( task_idlimited_task, python_callableprocess_color, pooldatabase_maintenance, )Pool 的配置路径在 UI 的 Admin Pools 中也可以在airflow.cfg中预定义。它比盲目调低全局 parallelism 更精细。8.5 设置合理的自动重试和 SLA对依赖外部系统的任务建议至少设置 2 次重试间隔 5 分钟以上。对必须准点产出的链路使用 SLA 监控和告警让运维在 SLA 超时前介入。8.6 不要用 XCom 传大数据XCom 适合传递任务参数、文件路径、行数统计等轻量数据不适合传 DataFrame、大 JSON。跨 Task 传大数据时先写对象存储或临时表再传递引用路径。8.7 DAG 代码可测试、可灰度DAG 文件是代码应该有基本的测试和评审流程。至少保证本地能解析、依赖关系经过 review、上线前在测试环境跑一次成功。对团队协作场景建议约定 DAG 命名规范、Task 命名规范、标签规范比如dag_id统一使用业务域加表名Task 统一用动词开头。9. 总结与后续学习方向回到最开始的问题。Airflow 看起来是一个“带颜色的调度界面”本质上却是一套完整的分布式工作流管理系统。它用 DAG 描述依赖用 TaskInstance 管理状态用 Executor 实现多种执行模式用 Web UI 提供可观测性。Graph View 里那一个个颜色块是这套系统最显性的价值体现把不可见的数据流变成了一眼可读的信号。这篇文章帮你完成了几件事弄清楚了 Airflow 解决什么问题掌握了 DAG、Task、Operator、Executor 等核心概念跑通了单机环境写出并运行了第一个动态生成任务的 DAG还梳理了生产环境部署和排错思路。下一步建议按照这个顺序继续深入第一把示例 DAG 改造成自己业务的简单管道跑通一个真实任务。第二用 docker-compose 部署一套 CeleryExecutor 环境感受分布式执行。第三阅读官方关于 DAG 依赖、时区、调度器底层的文档把时间规则彻底弄懂。第四结合团队业务设计 DAG 分层结构把 Airflow 接入生产数据链路并建立监控与告警机制。最后提醒一句Airflow 不是银弹。如果只是几个互不依赖的定时脚本cron 加一个像样的日志系统就能解决。但只要是任务之间存在依赖、需要精确控制重跑和回溯、需要多人协作维护的数据管道Airflow 会是值得长期投入的技术选型。
Apache Airflow调度系统实战：从DAG到分布式工作流
第一次打开 Airflow 的 Web UI绝大多数人的注意力都会停在 Graph View 那张五颜六色的图上。绿色代表任务成功红色代表失败蓝色代表运行中灰色代表排队等待还有被跳过、被上游阻断等特殊状态。很多人第一眼会觉得这界面有点“花里胡哨”而我的判断恰恰相反Airflow 之所以能从 Airbnb 内部工具成长为 Apache 顶级项目从“批量脚本调度器”变成数据平台事实标准的编排引擎最大功臣不是 DAG 语法也不是庞大的 Operator 生态而是这张“带颜色的图”背后的一整套任务状态机与调度语义。从工程本质看cron 和 Airflow 解决的问题完全不同。cron 只保证“到点执行一条命令”它不关心任务是否成功不知道任务之间是否有依赖也不能把失败批次挑出来重跑。Airflow 把数据管道从一张时间表变成一张有向无环图每个节点可以被观察、被重试、被并行调度、被限流而这些能力最终都汇聚在 UI 上那一个个颜色块里。所以题目中说 Airflow 是 built with Colors并不是在夸界面好看而是在说它把调度系统最核心的可观测性做成了默认能力。这篇文章会从痛点切入讲清楚 Airflow 的核心架构、核心概念、单机安装部署、第一个 DAG 的完整实现、生产环境改造思路、常见问题排查和最佳实践。文章示例基于 Airflow 2.x 编写3.x 的 DAG 定义方式整体保持一致具体安装版本请以官方发布为准。1. 这篇文章真正要解决的问题先看一个很常见的场景。某个数据团队维护着 30 多个定时任务任务是典型的“链路式”结构每天凌晨从业务库抽取数据经过清洗后写入数仓再触发指标计算最后给报表平台推送结果。最早的时候团队用 cron 写了一批脚本每个脚本负责一个环节。开始只有几个任务大家都觉得挺方便。随着任务变多问题开始集中爆发A 环节凌晨 2 点跑挂了B 环节 3 点照常启动读到的是不完整数据而且没有自动补偿。某天上游数据延迟到 4 点才到位下游脚本仍然严格按 cron 时间执行结果整条链路白跑。任务失败后运维要登录服务器手工重跑还要记住重跑命令时间一长就乱。想并行处理 6 个互不依赖的数据源cron 只能按固定时间串起来或者靠后台进程自己管理并发。老板问“今天的数据为什么没出”你只能看服务器日志没人能一眼说清失败发生在哪个环节。这些问题可以归纳为三个层次。第一层是依赖编排。任务之间不是孤立的A 成功之后才能跑 BB 失败后 C 不应该启动。cron 解决不了跨任务依赖只能靠脚本内部自己判断判断逻辑一旦分散就容易出现脏数据。第二层是状态管理。每个任务实例都应该有明确状态排队、运行中、成功、失败、跳过、被上游阻断。有了状态系统才能决定下一步动作才能支持失败重试、定时重跑、补历史数据。第三层是可观测性。调度系统不能只“执行”还必须让操作者随时看到当前工作流进行到哪一步。Graph View 里每个 Task 的颜色本质上就是把状态机可视化让“看日志猜原因”变成“看颜色定位问题”。Airflow 恰好在这三个层次上都提供了标准答案。如果你需要的是一个能处理跨任务依赖、支持失败重跑、能在分布式环境中扩展、并且有完整 Web UI 的开源调度平台Apache Airflow 是目前最值得优先考虑的选择。2. 核心概念DAG、Task、Operator 与 Executor在动手安装之前先把 Airflow 的核心概念讲清楚。很多初学者卡住不是卡在代码语法上而是没搞清楚对象之间的关系。2.1 DAG 是静态拓扑不是执行队列DAGDirected Acyclic Graph有向无环图是用 Python 代码描述的一整套工作流依赖关系。比如“先抽数、再清洗、再计算”就是三个节点和两条有向边。关键点在于DAG 本身只是一种拓扑描述它不会“执行任务”它只负责告诉 Scheduler“谁在谁前面”。这也解释了为什么 Airflow 的 DAG 文件是 Python 文件。Python 代码天然适合表达依赖关系而且可以在循环里动态生成 Task后面示例会展示这一点。2.2 Operator 是任务模板Task 是具体实例Operator 是 Airflow 预置或自定义的任务执行模板。BashOperator 用来执行 shell 命令PythonOperator 用来调用 Python 函数PostgresOperator 用来执行 SQLDockerOperator 用来运行 Docker 容器。你可以把 Operator 理解成“工具类”。同一个 Operator 在 DAG 中出现一次就生成一个 Task。比如一个 PythonOperator 模板在 DAG 里写了三个 Task它们功能相似但属于三个独立执行单元。再往下还有 TaskInstance 的概念它代表“某个 Task 在某个时间点的某次运行”。一个 Task 可以运行很多次每一次都是一个 TaskInstance每个 TaskInstance 都有独立状态。Graph View 里那个颜色块实际指示的是 TaskInstance 的状态。2.3 Executor 是真正干活的组件Executor 决定了 TaskInstance 用什么方式执行。这是 Airflow 架构里非常关键的一环。SequentialExecutor默认执行器同一时间只能跑一个任务适合本地调试生产环境不要用。LocalExecutor单机多进程执行并行度较高适合小规模部署。CeleryExecutor通过 Celery 分布式队列把任务分发给多个 Worker是生产环境最常见的方案。KubernetesExecutor每个 TaskInstance 动态创建一个 Kubernetes Pod 执行资源隔离好适合容器化平台。Scheduler 负责解析 DAG、判断哪些 Task 可以运行、把任务交给 ExecutorExecutor 负责真正调度执行。理解这个关系后面排查“任务一直 queued 不运行”时会非常有用。2.4 其他常用对象Sensor一种特殊的 Operator用来等待某个条件满足。比如等待文件生成、等待外部接口就绪。Hook封装外部系统连接的对象是 Operator 与外部系统交互的通道。XComTask 之间传递小消息的机制适合传递状态、路径、参数不适合传大数据。VariableAirflow 全局变量用于配置管理。Connection统一存储外部系统的连接信息包括主机、账号、密码避免在 DAG 代码里硬编码敏感信息。Triggerer从 Airflow 2.2 开始引入配合 Deferrable Operator 实现轻量级异步等待能显著降低空占资源的 Sensor 对 Scheduler 和 Worker 的压力。2.5 常见概念对比概念一句话理解容易混淆的点DAG工作流依赖图不是任务执行队列TaskOperator 在 DAG 中的实例不是具体某一趟运行TaskInstanceTask 在某一次调度中的运行同一 Task 可以有多个实例Operator可复用的执行模板不是运行实体Executor决定任务如何执行不是 Worker 本身Scheduler判断任务何时执行不真正执行业务逻辑XComTask 间传消息不适合传 DataFrame3. 环境准备与前置条件单机快速体验 Airflow比大多数人想象中简单。你只需要一个 Python 环境建议使用 Python 3.8 及以上版本。生产环境推荐 Linux PostgreSQL Redis但第一步只需要跑通本地。3.1 创建独立虚拟环境Airflow 依赖比较多强烈建议使用虚拟环境避免污染系统 Python。sudo apt update sudo apt install -y python3-venv python3-dev build-essential sudo mkdir -p /opt/airflow sudo chown -R $USER:$USER /opt/airflow python3 -m venv /opt/airflow/venv source /opt/airflow/venv/bin/activate注意如果你的服务器无法访问系统软件源可以只创建 Python 虚拟环境不安装系统依赖包。对大多数纯 Python 使用场景缺少编译工具也不影响 pip 安装。3.2 安装 Apache Airflow安装前设置 AIRFLOW_HOME这个目录会保存 airflow.cfg、日志和 SQLite 数据库文件。export AIRFLOW_HOME/opt/airflow pip install --upgrade pip pip install apache-airflow[celery,redis,postgres]这里安装了带 Celery、Redis 和 PostgreSQL 扩展支持的版本方便后面直接做生产化改造。如果只想快速体验只装apache-airflow也可以。安装完成后用命令确认版本airflow version3.3 初始化元数据库Airflow 默认使用 SQLite 作为元数据库。第一次使用要执行airflow db init这个命令会创建 Airflow 运行所需的表并生成airflow.cfg和webserver_config.py。执行完成后可以看到$AIRFLOW_HOME目录下出现 airflow.db 文件。3.4 创建管理员用户Airflow 2.x 默认启用 RBAC 认证必须创建用户才能登录 Web UI。airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com \ --password admin123生产环境一定要换成强密码并通过环境变量或外部密钥管理方式配置不要直接写在命令历史里。3.5 启动 Web Server 和 Scheduler新开两个终端分别启动两个核心进程。终端一airflow webserver -p 8080终端二airflow scheduler然后浏览器访问http://localhost:8080用刚才创建的账号登录。看到首页为空是很正常的因为还没有任何 DAG。接下来我们写第一个 DAG。4. 第一个 DAG彩虹批处理管道为了让示例贴近实际我用“颜色”作为业务维度构建一个批处理管道先准备数据再按颜色并行处理多个分片最后汇总并结束。这个 DAG 同时演示了动态生成 Task、参数传递、Jinja 模板和任务依赖。4.1 创建 DAG 文件在$AIRFLOW_HOME/dags目录下创建rainbow_pipeline.py。# 文件路径/opt/airflow/dags/rainbow_pipeline.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.empty import EmptyOperator from airflow.operators.python import PythonOperator COLORS [red, orange, yellow, green, blue, purple] def process_color(color: str, **context): 处理某个颜色对应的批次数据。 生产环境中这里可以替换为读取当日分区文件、调用下游接口、写入目标数据表。 ds context[ds] run_id context[run_id] print(frun_id{run_id}, ds{ds}, color{color}) print(fprocessing {color} batch ...) # 实际业务逻辑尽量保证幂等失败后可以安全重跑 default_args { owner: data_platform, depends_on_past: False, retries: 2, retry_delay: timedelta(minutes5), email_on_failure: False, email_on_retry: False, } with DAG( dag_idrainbow_pipeline, description一个用颜色展示 Airflow 状态语义与动态任务生成的最小 DAG, start_datedatetime(2024, 1, 1), scheduledaily, catchupFalse, max_active_runs3, tags[demo, colors], default_argsdefault_args, ) as dag: start EmptyOperator(task_idstart) prepare_data BashOperator( task_idprepare_data, bash_command( mkdir -p /tmp/airflow-demo/{{ ds }} echo source data ready /tmp/airflow-demo/{{ ds }}/source.txt ), ) color_tasks [] for color in COLORS: task PythonOperator( task_idfprocess_{color}, python_callableprocess_color, op_kwargs{color: color}, ) color_tasks.append(task) generate_summary BashOperator( task_idgenerate_summary, bash_command( echo all colors processed /tmp/airflow-demo/{{ ds }}/summary.txt ), ) end EmptyOperator(task_idend) start prepare_data color_tasks generate_summary end4.2 代码逻辑拆解这段代码有几个关键设计。在with DAG(...) as dag:块内创建的 Operator 会自动绑定到该 DAG所以不需要重复传dagdag。这种写法在 Airflow 2.x 中是官方推荐风格。COLORS列表配合for循环动态创建了 6 个process_red、process_orange、process_yellow、process_green、process_blue、process_purple任务。这在实际项目中非常常见业务维度事先不确定数据源是一个动态列表通过循环生成任务可以大幅减少重复代码。BashOperator 里的{{ ds }}是 Jinja 模板变量Airflow 在执行时会把ds替换成当前执行日期格式是YYYY-MM-DD。使用模板变量的好处是任务具备“运行时语义”同一个 Task 在每天的运行中会自动落在不同日期目录下。default_args中的retries和retry_delay是任务失败后的自动重试策略。对依赖外部系统稳定性的任务来说自动重试 1 到 3 次间隔几分钟能显著降低外部抖动导致的误报。4.3 验证 DAG 文件DAG 文件本质上是 Python 文件所以可以直接用 Python 解释器执行来检查语法python /opt/airflow/dags/rainbow_pipeline.py如果没有报错说明 DAG 可以正常解析。更严谨的做法是等 Scheduler 扫描到文件后用 Airflow 命令确认airflow dags list | grep rainbow airflow tasks list rainbow_pipelinetasks list会输出 DAG 中所有 Task 的名字正常情况应该包含start prepare_data process_red process_orange process_yellow process_green process_blue process_purple generate_summary end5. Web UI 操作与运行验证Airflow 最有价值的体验在 Web UI。登录后Pages 列表中会看到rainbow_pipeline。如果列表为空说明 DAG 文件没有被 Scheduler 扫描到这时先看airflow dags list-import-errors有没有导入错误。5.1 Graph View颜色状态图点击 DAG 名称进入 Graph View。这个页面是 Airflow 最经典的视图每个 Task 显示为一个节点节点之间的箭头表示依赖关系。未运行的任务默认是灰色。触发一次运行后节点颜色会随着生命周期变化蓝色任务正在运行。绿色任务成功。红色任务失败。橙色或黄色任务重试中。紫色任务被跳过。灰色任务排队等待。颜色的变化不是装饰而是状态机的可视化。当管道出现红色节点时你可以立刻看到失败发生在哪一环以及上游和下游受影响的范围。5.2 手动触发一次运行点击页面右上角的“播放”按钮或者使用命令行airflow dags trigger rainbow_pipeline触发后点进 Graph View会看到任务开始排队并陆续变成蓝色最终全部变绿。点击任意绿色节点可以查看该 Task 这一次运行的日志。我们可以在日志中看到process_color函数打印的内容。5.3 Grid View看总体运行情况Airflow 2.x 新增的 Grid View 把 DAG 每次运行和每个 Task 的状态放在一张表格里比 Tree View 更直观。这里可以快速确认每日运行是否成功、失败任务集中在哪里是日常巡检最常用的视图。5.4 触发条件与时间规则新手最容易困惑的是“为什么任务没有在我设定的时间运行”。Airflow 的时间规则有两点必须记住。第一start_date是第一个调度周期的起点不是第一次运行时间。对于scheduledaily第一个 Run 的触发时间通常是start_date 一个周期。比如start_datedatetime(2024, 1, 1)第一次运行会在 2024-01-02 凌晨触发而这一次运行的执行日期仍然是 2024-01-01。第二catchup默认是 True。如果start_date设置得比较早Scheduler 发现历史周期从未运行过会不断补跑历史任务。这在某些场景是优点但在大多数新开发场景中是惊吓。示例中显式设置catchupFalse避免产生一堆历史 Run。如果你发现任务“不在预期时间运行”优先检查三处start_date、schedule、时区配置。Airflow 默认使用 UTC如果你的业务在东八区且没有设置timezone凌晨 1 点的任务实际会在 UTC 凌晨 1 点运行也就是北京时间上午 9 点。6. 生产环境部署从单机到分布式单机跑通只是第一步。真实生产环境要考虑高可用、并发能力、数据安全性、日志管理等通常不会继续使用 SQLite 和 SequentialExecutor。6.1 推荐参考架构生产环境最常见的组合是PostgreSQL 存储元数据替代 SQLite。Redis 作为 Celery Broker负责任务消息传递。CeleryExecutor 负责任务分发。一个或多个 Worker 节点真正执行业务任务。Web Server 和 Scheduler 分离部署。可选部署 Flower用于监控 Celery Worker 状态。这个架构里Scheduler 只负责“决策”不负责“执行”。真正跑任务的是 Worker因此业务任务再重也不太会影响调度器稳定性。6.2 docker-compose 快速搭建如果团队已经使用 Docker用 docker-compose 搭建一套开发环境很快。下面是一个简化但可运行的参考。version: 3.8 x-airflow-common: airflow-common image: apache/airflow:2.10 environment: AIRFLOW__CORE__EXECUTOR: CeleryExecutor AIRFLOW__CORE__LOAD_EXAMPLES: false AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresqlpsycopg2://airflow:airflowpostgres:5432/airflow AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0 AIRFLOW__CELERY__RESULT_BACKEND: dbpostgresql://airflow:airflowpostgres:5432/airflow volumes: - ./dags:/opt/airflow/dags depends_on: - postgres - redis services: postgres: image: postgres:14 environment: POSTGRES_USER: airflow POSTGRES_PASSWORD: airflow POSTGRES_DB: airflow redis: image: redis:7 webserver: : *airflow-common ports: - 8080:8080 command: webserver scheduler: : *airflow-common command: scheduler worker: : *airflow-common command: celery worker说明几点。通过环境变量方式配置 Airflow是比直接改airflow.cfg更推荐的做法尤其是容器化场景。环境变量命名规则是AIRFLOW__SECTION__KEY例如AIRFLOW__CORE__EXECUTOR对应[core] executor。x-airflow-common是 YAML 锚点语法用于复用公共配置。实际部署时镜像 tag 请以官方可用版本为准不要直接拿生产环境使用latest。6.3 systemd 方式部署不依赖 Docker 的环境可以用 systemd 管理 Airflow 进程。以 Linux 用户airflow为例创建两个服务单元文件。/etc/systemd/system/airflow-webserver.service[Unit] DescriptionAirflow web server Afternetwork.target Wantsnetwork.target [Service] Typesimple Userairflow Groupairflow EnvironmentAIRFLOW_HOME/opt/airflow ExecStart/opt/airflow/venv/bin/airflow webserver -p 8080 Restarton-failure RestartSec10s [Install] WantedBymulti-user.target/etc/systemd/system/airflow-scheduler.service[Unit] DescriptionAirflow scheduler Afternetwork.target Wantsnetwork.target [Service] Typesimple Userairflow Groupairflow EnvironmentAIRFLOW_HOME/opt/airflow ExecStart/opt/airflow/venv/bin/airflow scheduler Restarton-failure RestartSec10s [Install] WantedBymulti-user.target启动并设置开机自启sudo systemctl daemon-reload sudo systemctl enable --now airflow-webserver sudo systemctl enable --now airflow-scheduler生产环境还应该配置日志定期清理、Worker 健康检查和元数据库备份。6.4 生产环境关键配置项以下配置项直接影响调度稳定性需要重点确认。配置项作用建议[core] executor指定执行器生产使用 CeleryExecutor 或 KubernetesExecutor[core] dags_folderDAG 文件目录固定路径避免误改[core] load_examples是否加载官方示例 DAG设置为 False[core] timezone全局时区按团队规范设置建议统一[core] parallelism全局最大并行任务数根据 Worker 资源调整[database] sql_alchemy_conn元数据库连接使用 PostgreSQL 并配置连接池[celery] broker_urlCelery 消息队列地址使用 Redis配置密码[celery] result_backendCelery 结果存储使用数据库后端[celery] worker_concurrency单个 Worker 并行任务数根据任务负载调优7. 常见问题与排查思路Airflow 使用中踩坑非常正常很多问题看起来千奇百怪根源就那么几类。整理几个高频问题供读者对照排查。问题现象可能原因排查方式解决方案Web UI 没有看到 DAGDAG 文件路径不对或 Python 语法错误airflow dags list-import-errors修复 DAG 文件确认dags_folder路径任务一直处于 queued不运行Executor 配置不对Worker 未启动或并发数占满查看 Scheduler 日志、Worker 状态确认使用 CeleryExecutor 并启动 Worker检查 parallelism任务运行时间不符合预期时区没设置或 start_date/schedule 理解有偏差查看 Run 的 execution_date统一时区配置按 Airflow 周期规则设计任务失败后自动重跑了一堆历史任务catchup 默认为 True查看 DAG 的 Run 列表新项目设置 catchupFalseDAG 文件刚修改UI 不刷新Scheduler 每分钟扫描一次 DAG 文件等待或手动重启 Scheduler了解 DAG 扫描机制避免频繁改 DAG调用外部服务报连接失败Connection 配置错误或密码被加密在 UI 的 Admin Connections 中测试正确配置 Connection确认 fernet_keyWorker 内存持续上涨任务并发过高或单个任务加载了太多数据用监控工具查看 Worker 资源调整 worker_concurrency优化任务代码部署后账号密码失效web server secret key 或 fernet_key 变化查看 Web UI 日志固定配置密钥备份 secrets7.1 DAG 导入错误是最高频问题Airflow 的 Scheduler 会周期性扫描 DAG 文件目录。只要文件里有未捕获的异常整个文件对应的 DAG 都会被标记为导入失败。排查时第一时间执行airflow dags list-import-errors如果输出里有错误信息按照 Python traceback 处理即可。常见原因包括顶层代码访问了不存在的环境变量、依赖包未安装、文件里写了与 Airflow 无关的启动逻辑。7.2 任务排队不执行任务一直 queued是生产环境最常见的“假死”现象。在 CeleryExecutor 环境下先检查 Worker 进程是否存在ps -ef | grep celery worker再检查[celery] worker_concurrency是否被占满。一个 Worker 的并发数是有限的如果所有并发槽都被长任务占满新任务就会一直排队。此时需要增加 Worker 节点或者调优任务执行时间。7.3 日志和元数据越来越庞大Airflow 运行时间越长日志和元数据表增长越快。建议定期清理历史日志并考虑将日志接入集中式日志平台。元数据表里的 TaskInstance 数据也可以按保留策略清理避免数据库膨胀拖慢 Scheduler 扫描。8. 最佳实践与工程建议要把 Airflow 在团队里用好光会写 DAG 还不够。以下建议来自大量生产实践基本是按优先级排序的。8.1 任务必须幂等这是所有数据管道的第一原则。同一个 Task 无论重跑多少次结果应该一致。典型做法是写入数据前先清理目标分区再写入当日新数据文件任务先写临时目录成功后再移动到最终目录。幂等设计能保证失败重跑、补数、修复数据时不会产生脏数据。8.2 一个 Task 只做一件事把一个大任务拆成多个小任务会增加调度开销但收益更大失败定位更精确重试粒度更小可读性更好。理想情况下一个 Task 失败后单独重跑它就能恢复不需要重跑整条链路。8.3 敏感信息不硬编码数据库密码、接口密钥、云厂商凭证不应该出现在 DAG 文件里也不应该写在default_args中。Airflow 的 Connections 和 Variables 就是为此设计的。生产环境建议配合外部密钥管理服务使用。8.4 用 Pool 控制并发资源如果多个 DAG 共享同一套资源比如同一个数据库集群使用 Pool 可以限制同时运行的任务数。from airflow.utils.state import State from airflow.operators.python import PythonOperator task PythonOperator( task_idlimited_task, python_callableprocess_color, pooldatabase_maintenance, )Pool 的配置路径在 UI 的 Admin Pools 中也可以在airflow.cfg中预定义。它比盲目调低全局 parallelism 更精细。8.5 设置合理的自动重试和 SLA对依赖外部系统的任务建议至少设置 2 次重试间隔 5 分钟以上。对必须准点产出的链路使用 SLA 监控和告警让运维在 SLA 超时前介入。8.6 不要用 XCom 传大数据XCom 适合传递任务参数、文件路径、行数统计等轻量数据不适合传 DataFrame、大 JSON。跨 Task 传大数据时先写对象存储或临时表再传递引用路径。8.7 DAG 代码可测试、可灰度DAG 文件是代码应该有基本的测试和评审流程。至少保证本地能解析、依赖关系经过 review、上线前在测试环境跑一次成功。对团队协作场景建议约定 DAG 命名规范、Task 命名规范、标签规范比如dag_id统一使用业务域加表名Task 统一用动词开头。9. 总结与后续学习方向回到最开始的问题。Airflow 看起来是一个“带颜色的调度界面”本质上却是一套完整的分布式工作流管理系统。它用 DAG 描述依赖用 TaskInstance 管理状态用 Executor 实现多种执行模式用 Web UI 提供可观测性。Graph View 里那一个个颜色块是这套系统最显性的价值体现把不可见的数据流变成了一眼可读的信号。这篇文章帮你完成了几件事弄清楚了 Airflow 解决什么问题掌握了 DAG、Task、Operator、Executor 等核心概念跑通了单机环境写出并运行了第一个动态生成任务的 DAG还梳理了生产环境部署和排错思路。下一步建议按照这个顺序继续深入第一把示例 DAG 改造成自己业务的简单管道跑通一个真实任务。第二用 docker-compose 部署一套 CeleryExecutor 环境感受分布式执行。第三阅读官方关于 DAG 依赖、时区、调度器底层的文档把时间规则彻底弄懂。第四结合团队业务设计 DAG 分层结构把 Airflow 接入生产数据链路并建立监控与告警机制。最后提醒一句Airflow 不是银弹。如果只是几个互不依赖的定时脚本cron 加一个像样的日志系统就能解决。但只要是任务之间存在依赖、需要精确控制重跑和回溯、需要多人协作维护的数据管道Airflow 会是值得长期投入的技术选型。