企业官网建设流程全解析

1. 为什么我最终把核心调度从Cron和脚本升级成了Airflow

先交代一下背景。前几年我在某公司的数据组维护着将近二十套批处理脚本，有Python写的，有Shell套SQL的，还有几段Java程序靠Cron触发。表面上一切正常，实际上每次出问题都是连环爆炸：A表凌晨两点算完，B表三点依赖它，结果A任务因为上游接口超时挂了，B脚本拿到的是昨天的分区数据，C报表跟着把脏数报给了业务线。那时候排查问题全靠人工对时间轴，群里永远有人在问"昨晚的任务到底跑没跑"。

后来我把这套东西整体迁移到了Apache Airflow上，用DAG把每个任务的依赖关系明确写出来，失败自动重试，重跑补数一条命令搞定，调度状态在Web界面上一目了然。可以说，对于定时批处理管道，Airflow解决的核心问题不是"定时触发"——Cron也能定时——而是三个Cron和普通脚本给不了的能力：依赖编排、失败重试与数据回填、可视化态势感知。

这篇文章不是Airflow官方文档的复述，我会按自己从零搭建到生产落地的完整路线来讲，包括DAG和Task这些核心概念到底什么意思、调度时间为什么经常踩坑、开发环境怎么用Docker Compose快速搭、生产管道设计时要注意哪些反模式，以及最后哪些场景其实不该用Airflow。写给正在数据工程这条路上爬坡的同学，也写给被Cron脚本折磨到想重构的老手。

2. 先弄明白DAG、Task、Operator和Scheduler之间的分工

2.1 DAG是"流程蓝图"，Task Instance才是"一次具体运行"

很多人刚接触Airflow时会被一堆名词绕晕。我用一句话帮你理清：DAG描述的是管道长什么样，Task描述的是管道里有哪些活，Task Instance描述的是某一次实际跑起来的某个活。

DAG全称Directed Acyclic Graph，有向无环图。有向意味着每个任务都有明确的前后指向，A完成后才轮到B；无环意味着不允许出现"任务A等任务B，任务B又等任务A"的死循环。Airflow把你这套数据处理流程画成一张图，这张图本身是静态定义，用Python代码写出来。比如你每天凌晨同步订单数据、清洗、聚合、写结果表，这四步就是一个四条节点的DAG。

Task是DAG里的单个节点，可以理解成"一个最小工作单元"。它对应一个Operator实例，比如BashOperator表示跑一条Shell命令，PythonOperator表示执行一个Python函数，这是Airflow最常用的两种。

那Task Instance就是某个时刻某个任务的真实执行记录。同一个Task可以跑很多次：每天跑一次，跑失败了重试一次，后来数据出问题又手动触发一次，每一次都是一条Task Instance。在Web界面里你看到的每条色块记录，就是Task Instance。

2.2 Scheduler、Executor、Worker：调度引擎三个人

Airflow在2.x版本常被描述为"一个以元数据数据库为中心的调度平台"，这个说法很准确。它的核心组件有四个：

这里有个新手常见误区：Scheduler不是说"时间一到就立刻执行一个任务"，它更像一个不停扫描的时刻表。每隔几秒（默认大约30秒）Scheduler是否准备就绪：依赖是否满足、上一个实例是否成功、是否到了触发时间。所以你的DAG改了代码，要等一个扫描周期才会反映出来，这很正常。

2.3 Operator是"干活零件"，Sensor是"等等再干活"的零件

Operator的种类非常多，我没有全部用过，但实践中90%的场景集中在下面几个：

• BashOperator：直接跑Shell命令，比如调用SQL命令行、执行Python脚本
• PythonOperator：调用Python可调用对象，适合写业务逻辑
• DummyOperator：什么都不干，常用于串联依赖关系或画图分组
• ExternalTaskSensor：等待另一个DAG的某个任务完成再继续，用于跨管道编排
• SQLExecuteQueryOperator：直接执行SQL语句，适合对接数仓

Sensor是一类特殊的Operator，它的任务是"等待某个条件成立"。比如上游文件系统里某个日期的文件夹还没出现，Sensor会周期性轮询，直到文件出现才进入成功状态，后续任务才开始跑。这种机制可以优雅地化解"上游系统没有准点产出"的问题，而不是傻傻地在某个固定时间硬跑。

3. 本地开发环境搭建：纯手工安装还是Docker Compose？

3.1 为什么我推荐Docker Compose而不是直接pip install

Airflow的依赖很多，和系统Python环境、数据库驱动版本相互影响，直接pip install到宿主机很容易遇到"装好了却起不来"的问题。我踩过的坑包括：Python版本不匹配导致C扩展编译失败、找不到配置目录、Scheduler启动后反复报数据库初始化错误。后来统一改成Docker Compose，十分钟就能拉起一套可用的环境，还能随意切换版本。

一个标准的docker-compose.yaml核心技术点如下：包含五个服务——scheduler、webserver、worker、triggerer（2.x以后异步任务需要）、以及用于初始化数据库的一次性服务。Airflow官方提供了生产级别的docker-compose.yaml，直接下载后用，但我建议你修改两处：一是把默认的AIRFLOW__CORE__LOAD_EXAMPLES设为false，避免每次打开Web界面都被一堆示例DAG刷屏；二是把FERNET_KEY换成自己生成的值，这个密钥用于加密连接密码，不换的话虽然能用，但属于安全隐患。

3.2 编写第一个DAG：从代码逐行读懂调度逻辑

我们用Airflow 2.x的语法写一个最简单的"每分钟打印时间"DAG，你先有个体感，后面再深入细节。

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator default_args = { "owner": "data_team", "depends_on_past": False, "retries": 3, "retry_delay": timedelta(minutes=5), } with DAG( dag_id="simple_print_dag", default_args=default_args, description="最简单的每分钟打印时间示例", schedule_interval="*/1 * * * *", start_date=datetime(2024, 1, 1), catchup=False, tags=["example"], ) as dag: def print_current_time(): print("当前任务执行时间") print_task = PythonOperator( task_id="print_time", python_callable=print_current_time, ) print_task

在这个DAG里有几个字段非常关键，很多人就是没搞懂它们之间的关系而出问题：

• schedule_interval：定时规则，用Cron表达式或timedelta。注意Airflow 2.x推荐用schedule字段，但schedule_interval仍然兼容
• start_date：调度开始日期，它代表的是"第一个调度周期从什么时候开始"，而不是"任务第一次在哪一秒跑"
• catchup：是否补跑缺失的历史调度。设为False时，如果你今天才创建了一个start_date在三个月前的DAG，它不会把过去三个月的几十次调度都一口气补跑

DAG文件写好后放到dags目录下，Scheduler会自动识别并解析。你可以在Web界面看到这张图：一个节点，没有任何连线，因为没有依赖。

3.3 为什么说"代码即配置"是一把双刃剑

Airflow用Python代码定义DAG，好处是灵活：可以在循环里批量生成一百个相似的Task，可以用if判断动态决定管道走向，可以import项目里的公共工具函数。但坏处是Python代码的自由度也带来了难以察觉的错误。

最典型的是"顶层代码导入慢"问题。Scheduler每隔几十秒会去解析DAG文件，如果你的文件顶部有import pymysql这种重库，或者有连接数据库获取配置的顶层代码，每次解析都会被拖慢，严重时Scheduler会判定超时并雪崩式报错。我在生产环境见过有人把requests.post写进顶层变量初始化，结果每一次Scheduler心跳都触发一次外呼请求，直接把内部接口打到限流。

所以规范是：顶层只做轻量导入和常量定义，任何耗时操作都放进任务函数里。这是"代码即配置"模式下必须养成的习惯。

4. 调度时间与回填逻辑：为什么任务总比我以为的晚跑一小时

4.1 Data Interval机制：这个设计让很多人一开始很困惑

Airflow 2.x引入了一个非常核心的概念：Data Interval（数据区间）。每个DAG Run都对应一段数据时间区间，任务处理的是这个区间内的数据，而任务实例的开始时间在区间结束之后。

举个例子：你设置schedule_interval="0 2 * * *"，代表每天凌晨2点触发，start_date设为1月1日。那么1月1日对应的DAG Run，它处理的是1月1日这一天（00:00到24:00）的数据，而它实际的执行时间却是1月2日凌晨2点。如果你用task_id >> task_id写依赖，上游1月1日的DAG Run跑完后，下游1月1日的DAG Run才开始，它们一直在同一个数据区间内对齐，不会出现"下游拿到昨天数据"的错位。

很多人手动测试时会用"Run"按钮点击”Trigger DAG”，然后看到DAG Run的logical_date（逻辑日期，2.x版本后改名，以前叫execution_date）是昨天的日期，会以为自己搞错了。其实这恰恰是Airflow的正常设计：手动触发时会默认使用当前数据区间的开始时间作为逻辑日期，而你实际等的是这个区间结束后才开始执行。

4.2 catchup=True时的"补跑雪崩"问题

catchup是另一个高频踩坑点。很多教程里都会写catchup=False，但没人告诉你为什么。

假设你上周五才创建了一个DAG，start_date设为上周一，schedule_interval是每天一次。catchup=True会让Airflow从上个周一到现在每个缺失的调度周期各生成一个DAG Run。如果中间隔了五天，就是五个Run同时或先后启动。如果你的任务不支持并发写同一张表，就可能出现数据错乱、重复写入、共享临时文件互相覆盖。

我在测试环境遇到过更夸张的：一个原本每小时跑一次的DAG，上线两周后才发现start_date设置错了，改回去后catchup=True一口气生成了几百个Run，把整个Executor的队列塞满，正常的实时任务全被挤到后面排队，搞了两小时才缓过来。

所以我的建议默认是：新上线的DAG，catchup要么False，要么把start_date设成当天零点。只有确定要做历史数据回填时才开True，并且回填完成后立刻再改回False。

4.3 depends_on_past与任务级串行

如果你希望同一Task在上一次实例成功之前不运行下一次，可以在default_args里设置depends_on_past=True。这个参数适合有严格顺序依赖的场景，比如"每天的全量快照表，今天的量必须依赖昨天跑成功，否则基础数据链断裂"。

但要注意一个反直觉的点：depends_on_past=True会让失败恢复变得麻烦。假设昨天的任务失败了，你今天把代码修好，想重跑今天的任务，Airflow会因为昨天的实例失败而拒绝启动今天的实例。你必须先把昨天的失败任务手动重跑或者Mark Success，今天的才会开跑。这个机制上有它的合理性，但生产环境里容易造成"卡死循环"。

5. 实战设计一个订单数仓管道：多任务编排与依赖写法

5.1 一个贴近真实场景的DAG设计

我拿一个最常见的场景举例：每天从业务库抽取订单数据，写入数仓ODS层，再清洗到DWD层，最后做汇总指标进DWS层，并发送数据质量报告邮件。四个层级，四条依赖线。

用Airflow实现，我会这样组织：

from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator from airflow.utils.trigger_rule import TriggerRule with DAG( dag_id="order_etl_pipeline", schedule_interval="0 1 * * *", start_date=datetime(2024, 3, 1), catchup=False, default_args=default_args, ) as dag: # 1. 抽取订单数据到ODS层 extract_orders = BashOperator( task_id="extract_orders_to_ods", bash_command="python /data/scripts/extract_orders.py --ds {{ ds }}", ) # 2. 清洗订单明细到DWD层 clean_orders = BashOperator( task_id="clean_orders_to_dwd", bash_command="python /data/scripts/clean_orders.py --ds {{ ds }}", ) # 3. 聚合指标到DWS层 aggregate_metrics = BashOperator( task_id="aggregate_metrics_to_dws", bash_command="python /data/scripts/aggregate_metrics.py --ds {{ ds }}", ) # 4. 发送质量报告 send_report = PythonOperator( task_id="send_data_quality_report", python_callable=send_report_func, trigger_rule=TriggerRule.ALL_DONE, ) extract_orders >> clean_orders >> aggregate_metrics >> send_report

这里{{ ds }}是Jinja模板变量，会被解析为当前DAG Run的数据区间开始日期（格式YYYY-MM-DD）。通过它传给脚本，脚本就能精确地只处理对应日期的数据，这是Airflow实现"按分区处理"的标准做法。

5.2 三种依赖关系写法与选择

• 位移运算符：task1 >> task2，直观对标"先A后B"
• set_upstream/set_downstream：适合在循环里动态设置，或者任务特别多时程序化控制
• list方式：task1 >> [task2, task3] >> task4，表示task2和task3并行执行，都完成后才执行task4

养成一个习惯：能用位移运算符就表意优先，因为它读起来最像口语："先把数据抽取出来，再清洗，再聚合"。在复杂的菱形依赖中，配合列表写并行任务，可读性依然很高。

5.3 跨DAG依赖用ExternalTaskSensor还是TriggerDagRunOperator

单管道内用位移运算符就行了，但真实业务经常需要跨管道依赖，比如"日志管道跑完，订单管道才能开始跑"。

我有两种选择：

1. ExternalTaskSensor：在订单管道里挂一个Sensor，死死等待日志管道某个任务在某次DAG Run中成功
2. TriggerDagRunOperator：日志管道跑完后主动触发订单管道

我的经验是：Sensor方案更符合"数据就绪才跑"的理念，但配置不当容易死锁。Sensor需要指定外部DAG的external_dag_id、external_task_id和execution_delta。最经典的死锁场景是：外部DAG因为某种原因压根没在这个数据区间生成DAG Run，Sensor会一直等到超时，而订单管道里排在Sensor后面的任务全部被堵住，资源被占着，后续调度越来越多堆积。

我自己的处理方法是：给所有Sensor设置timeout和sla，比如20分钟后没等到就失败并通知值班群，让告警直接拉人介入，而不是埋在任务队列里假死。另外，优先保证上游管道是稳定的、有时间保障的，才用跨管道Sensor。上游本身就是"可能不准时"的外部系统，建议用文件级Sensor去轮询像HDFS文件路径这类的就绪信号，而不是跨DAG Run去等。

6. 生产环境里那些没人写进文档的失败教训

6.1 最大的坑：任务不幂等，回填数据翻车

Airflow的定位就是可以随时重跑任意时间范围的任务。这个能力非常强大——前提是任务本身要幂等。

什么叫幂等？同一个任务，同样参数，重跑十次，结果都一样。我在项目里部署的第一个生产管道就栽过：清洗任务会在目标表里用"先删后插"的方式写当日数据，看起来没问题，可一旦重跑同一天的任务，它会先删掉昨天的数据再插今天的。当时上游数据源恰好把昨天某个字段值修正了，我重跑今天凌晨的任务时，Date条件用的是模板变量--ds 2024-03-02，结果把3月2日的数据重写了，但3月1日的旧数据还留着，两边数值对不上。

所以构建管道时，每个写数据的目标表都必须有明确的分区标识或唯一键，任务里要做的事是"先清该分区再写该分区"或"upsert"，而不是全表覆盖。你可以在脚本开头打印一段"任务开始处理的数据区间"来确认参数传值没传错，这招在排查问题时特别有用。

6.2 并发与资源控制：一个Worker被拖死全队遭殃

Airflow默认允许一个DAG同时跑很多个Task Instance，如果你在代码里写了个死循环或者某个任务申请了大量内存，很可能把Worker节点拖垮。

生产环境我建议在DAG级别设置max_active_runs=1（同一时间只允许一个DAG Run在跑），在任务级别使用pool字段把不同任务放进不同资源池，限制并发数。

例如"普通ETL任务"这个池只能同时跑10个，"重资源机器学习任务"这个池只能同时跑2个：

# 在airflow.cfg或环境变量中配置Pool AIRFLOW__CORE__DEFAULT_POOL=default_pool # 代码中指定 heavy_task = BashOperator( task_id="heavy_ml_training", bash_command="python train_model.py", pool="ml_pool", )

Pool需要先在Web界面Admin -> Pools中创建，然后代码里引用。并发控制做得好不好，直接决定一个故障是"局部任务失败"还是"整个调度平台雪崩"。

6.3 时区问题：一天的数据等到第二天12点才算完

Airflow支持时区配置，默认是UTC。这和国内业务常用的东八区存在八小时偏差。

我踩过一次很隐蔽的时区坑：DAG设置的schedule_interval是0 2 * * *，我想的是北京时间凌晨2点跑，但Airflow按UTC算，实际上是北京时间上午10点才跑，下游每天10点后收到数据，业务方等得焦头烂额，我却一直没看日志里的时间戳。

排查方式：进入Web界面看DAG Run的logical_date和start_date，一对比就露馅。解决方案是把AIRFLOW__CORE__DEFAULT_TIMEZONE设置成Asia/Shanghai，同时确保宿主机和Docker容器的时区一致。但注意：start_date里写的datetime()如果不带时区信息，Airflow会默认使用你配置的默认时区，所以显式带上timezone("Asia/Shanghai")会更保险。

6.4 日志查看：tty宽度与日志丢失

Airflow的日志存在Worker本地磁盘上，如果你用CeleryExecutor跑分布式，日志在每台Worker上，Web界面通过读取远端文件来展示。这个机制在小集群上没问题，但日志量大、Scheduler与Worker文件系统不一致时，Web界面经常出现"日志读取失败"。

我的建议是尽快接上远程日志存储，比如S3或阿里云OSS，Airflow原生支持远程日志处理器，配置好remote_log_conn_id和remote_base_log_folder后，任务日志能统一归档到对象存储。否则你排查生产问题时会到处翻Worker机器，效率极低。

7. 调度管道的检测与告警配置

7.1 内建告警：SLA和email_alert

Airflow可以在任务失败时自动发送邮件。配置方式是在default_args里设置email_on_failure=True和email列表。但这有个前提：Web服务器环境要能发出邮件，需要在airflow.cfg中配置SMTP连接。我自己在内部运维体系中很少走邮件，更习惯通过on_failure_callback挂一个回调，把失败信息POST到企业内部的告警接口。

def send_failure_alert(context): dag_name = context["dag"].dag_id task_name = context["task"].task_id alert_payload = {"dag": dag_name, "task": task_name} requests.post("http://internal-alert-center/api/alert", json=alert_payload) default_args = { "on_failure_callback": send_failure_alert, }

7.2 态势感知比告警更重要

如果每个任务失败都告警，一段时间后团队会对告警麻木。我更推荐在DAG顶端加一个"态势感知任务"，它不处理具体数据，而是汇总下游结果。这样调度平台里每天会有两个核心节点：所有数据任务结束后运行的sla_report_task，以及凌晨统一检查各表数据最新分区的data_quality_check_task，后者用SQL查每个关键表的max(partition_date)，如果发现某张表今天没更新，立刻发升级告警。

这种设计让我从"等告警再排查"变成"每天固定时间主动看全局"，实际运维体验好很多。

8. Airflow的边界：什么场景不要用Airflow

Airflow被广泛用于批处理编排，但它不是一个实时流处理系统。如果你在纠结"我该不该用Airflow做实时管道"，答案是：大概率不该。

8.1 流式任务和秒级任务

Airflow的最小调度粒度受限于Scheduler扫描周期，秒级任务会把Scheduler拖得疲惫不堪。你要做实时流处理、秒级监控，应该用Flink或Spark Streaming这类流式计算引擎，Airflow适合做这些框架的"日常守护神"：定期启动它们、检查状态、失败重拉。

8.2 单体脚本里只有一个Python函数

如果一个任务几十行就搞定了，就一个Python脚本，不需要多任务依赖，也不要求可视化调度界面，那直接Cron就好。Airflow最核心的价值在于帮你理清复杂的依赖关系和运行历史，当管道里任务数量少于三个、依赖只有一条线，引入Airflow反而增加开发和运维成本。

8.3 依赖在运行时才知道的动态DAG

Airflow的设计哲学讲究"静态DAG，动态运行"，意思是DAG的结构应当在加载时确定下来。虽然你可以用动态生成方式根据外部配置创建几百个几乎相同的DAG，一旦配置变化就重新生成DAG，这个玩法在任务数量有限时没问题，但如果你的管道依赖是从上游API动态获取的——比如用户传来的任意拓扑结构——Airflow的调度模型会很吃力。这种场景更适合用专门的编排系统或工作流引擎来处理。

我最后想说的几句经验

我这两年从Cron脚本迁移到Airflow，最大的感受不是工具本身多强大，而是它推着你把管道设计得规范：每个任务必须有清晰的幂等边界，依赖关系要画在明面上，失败重跑要有操作的入口，产出要有质量检查节点。这些工程化习惯，比Airflow任何一项具体功能都重要。

如果你正准备在团队里引入Airflow，我建议先从一两个非核心管道开始试点，把前面说的调度时间、时区、并发、幂等这几块全部走通，再逐步把核心管线迁过来。别一上来就追求CNCF里的花式功能，先把DAG写好，把Web界面看熟，把日志和告警跑通，这套基础功比什么都值钱。

如果大家在实际搭建过程中遇到什么奇怪的问题，欢迎在评论区留言，我尽量把当年踩过的坑一个个回给你们。