凌晨一点电话响了。某零售平台每日销售报表没出我爬起来看日志发现清洗脚本跑挂了而依赖它的聚合任务没有执行前端看板整整空了一上午。这种“管道事故”做过数据的人都懂脚本一个个串起来没人关心依赖顺序失败了没人重试重试了又没人通知。后来我接手这类活第一件事就是把所有定时脚本迁到Apache Airflow里用DAG把它们编排成看得见、管得住的数据管道。Apache Airflow是目前数据工程领域最常用的一套开源工作流编排平台核心思路是用Python定义一组有向无环图把“任务是什么”“谁依赖谁”“什么时候跑”“失败了怎么办”全部写进代码里。它解决的是批处理场景下的调度、依赖、重试、监控和日志问题适合数据工程师、数仓开发、后端兼职数据管道的人也很适合那些已经用crontab跑了几十个脚本、出了事只能靠人肉盯的团队。想用好Airflow首先要建立两个认识管道不是脚本的简单串行调度不是cron的加强版。下面我从选型逻辑、组件部署、DAG写法到生产坑位一条线讲清楚。1. 为什么偏偏是Airflow工作流编排的底层逻辑1.1 数据管道从来不只是一堆脚本的串行很多人做数据管道的起点是crontab。每天两点跑抽取三点跑清洗四点半跑聚合六点给报表发邮件。听起来没问题但真实场景很快会失控清洗任务因为上游数据晚到而失败聚合任务却已经按时启动了等发现的时候整个链路可能已经跑了两个小时或者某个任务明明失败了因为cron不会自动重试第二天早上老板看到的还是昨天的数据。我把这种模式比作后厨备菜洗菜、切菜、炒菜是严格串行的小流程任何一个环节出问题后面所有环节都得停。手工备菜可以等但自动调度不会“等”它只会机械地到点执行。这就是cron的天然短板它只管“到点触发”不管“是否应该触发”也不管“前置是否成功”。Airflow的DAG恰好把“前置条件”这种依赖关系做成了第一等公民每个任务都必须等上游状态满足条件才启动。做了几年数据管道我最深的感受是脚本本身通常不复杂复杂的是脚本之间的依赖、失败后的恢复顺序以及跨天跨周的数据重算逻辑。Airflow把这些问题变成了一套可描述的代码和可观察的状态这比脚本本身值钱得多。1.2 DAG思维把管道写进代码里DAG是有向无环图Airflow里的每个任务是一个节点任务之间的依赖是边。与脚本最大的区别是Airflow把一次运行的完整状态“物化”到了系统里每个任务是什么状态、成功还是失败、跑了多长时间、输出过什么日志全部有据可查。调度历史也变成了一张清晰的运行时间线出问题的时候能快速回到现场。用代码而不是YAML或者界面拖拽来描述管道是因为现实管道里经常有判断、分支、循环还有基于参数的动态逻辑。Python表达能力足够强而且同一套DAG代码可以进Git评审、可以多人协作代码Review能看出潜在问题。我的习惯是每个DAG文件控制在几百行以内反复出现的逻辑抽成helper函数这样读起来舒服也方便做权限和归属管理。有一点务必注意DAG代码只是“描述”不是“执行”。每次Scheduler扫描到文件都会解析成一个对象图再与元数据库里的历史状态比对决定是否需要生成新的任务实例。所以DAG文件里不应该写真正的业务计算逻辑那部分交给任务内部的Python函数或Shell脚本避免解析文件时白白消耗资源。1.3 该不该用Airflow先想清楚场景Airflow虽然强大也不是银弹。如果只是三五个脚本用cron加一个简单的失败告警就够了引入调度平台反而增加运维成本。如果业务是流式的秒级低延迟场景Airflow的批处理调度模型也不适合那是流处理框架的领域。Airflow真正舒服的区间是“小时级、天级的批处理管道任务数量几十到几千依赖关系复杂、需要重试和监控”的场景。我见过很多团队一上来就上KubernetesExecutor几十个Pod跑几个轻量任务折腾半天收益很低。方案选型必须跟数据量、任务量、团队维护成本匹配否则Airflow就会变成比脚本更复杂的坑。拿我自己的经验来说一个中型数仓团队每天跑三五百个任务用CeleryExecutor配三五台Worker已经很宽裕完全不需要一开始就把基础设施整得很重。另外还要想清楚团队能投入的学习成本。Airflow的张牙舞爪其实很浅入门一周足够但如果团队连“虚拟环境”“代码仓库”都用不熟练直接上Airflow只会放大混乱。工具的目的是补人的短板不是替代人的规范。2. 先把地基打好Airflow核心组件与部署选型2.1 四个角色和一条元数据库理解Airflow的运行时架构抽象成四个角色就够了Scheduler调度器负责解析DAG文件、根据时间生成任务实例然后交给执行器排队。它是整个系统的大脑。Worker真正执行任务的地方默认会在Scheduler所在进程里跑生产环境通常独立成多台机器。WebserverWeb UI提供可视化界面看DAG、触发任务、看日志。Metadata Database元数据库保存DAG定义、任务实例状态、调度记录、执行结果。很多人第一次接触会混淆Scheduler和Worker。可以这么记Scheduler只做“决定”决定哪个任务现在该跑Worker只做“执行”把决定变成真正的代码运行。元数据库是粘合剂所有状态都写在那里Scheduler、Worker、Webserver靠它同步信息。生产中如果条件允许我会把Scheduler和Webserver分开跑数据库用PostgreSQL而不是SQLite。Scheduler高负载时会频繁读写数据库SQLite在并发写入下很容易出现锁等待轻则调度变慢重则元数据库损坏。用PostgreSQL之后这种问题基本绝迹。2.2 执行器怎么选LocalExecutor还是CeleryExecutor执行器负责把待运行的任务真正跑起来。单机安装默认的SequentialExecutor必须依赖SQLite同一时刻只能跑一个任务只适合学习和演示千万别拿到生产用。小规模场景每天几十个任务我建议直接用LocalExecutor它靠多进程并发执行任务部署简单稳定可靠跑两百个任务都不会有明显压力。等任务量上来、单机CPU和内存吃不住了再切CeleryExecutor多台Worker并行通过Redis或RabbitMQ分发任务。切换时核心变化是worker组件从单进程变成多节点DAG代码需要同步部署到每一台Worker节点上。任务需要特定依赖环境的还可以按队列分发到不同worker池比如把重任务和轻任务分流。实际项目里我习惯先把单机LocalExecutor跑到极限再说盲目上分布式只会让排查问题变得复杂。Airflow的调度问题更多出现在调度配置和时间规则上而不是执行器本身。2.3 部署方式与踩坑记录最快起步方式是直接跑官方提供的docker-compose约等于把Webserver、Scheduler、Worker、数据库、Redis那套一次拉起。第一次启动前先在环境变量里配置强随机的FERNET_KEYAirflow用它加密连接串默认的示例密钥相当于家门没锁任何人用默认的弱密钥来试都是危险的。启动后本机访问8080端口先创建一个管理员账号再把环境自带的所有示例DAG停掉免得一堆无用任务干扰排障。不想用容器时本地直接pip安装也能跑。建议用虚拟环境隔离避免和系统Python里其他包冲突。配置时重点盯几个地方AIRFLOW_HOME指向的目录要有写权限airflow.cfg里的timezone要设成Asia/Shanghai数据库连接串协同PostgreSQL使用时要带时区参数否则时间统计会乱。我第一次装的时候在时区上吃过亏后面专门讲。3. 从零写第一个DAG把日常脚本管起来3.1 最简DAG长什么样下面是一个最基础的DAG每天凌晨两点执行一次先跑一个Python函数再跑一段Shell脚本from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator default_args { owner: data_team, depends_on_past: False, start_date: datetime(2025, 1, 1), retries: 1, retry_delay: timedelta(minutes5), } with DAG( dag_iddemo_daily_pipeline, default_argsdefault_args, schedule_interval0 2 * * *, catchupFalse, tags[demo], ) as dag: def extract(): print(开始抽取数据) # 这里写实际抽取逻辑 return done task_extract PythonOperator( task_idextract_data, python_callableextract, ) task_clean BashOperator( task_idclean_and_load, bash_commandbash /opt/scripts/clean_and_load.sh, ) task_extract task_cleandefault_args用来统一设置owner、重试次数和间隔不需要每个任务都重复。schedule_interval接受cron表达式这种每天两点的表达大家都不陌生。catchupFalse特别重要它决定启动DAG时是否回补历史时间点的运行记录新测试的DAG建议先设False否则很容易一口气把过去几个月的任务全部触发一遍直接把资源池打满。3.2 依赖关系不是只有“上一个成功”任务编排支持多种依赖方式task_a task_b表示B依赖AA成功才跑B复杂一点可以用task_a [task_b, task_c]让A并行触发两个下游再用[task_b, task_c] task_d等它们都完成再跑D。更细的控制在触发规则上默认all_success要求所有上游成功后当前任务才执行。实际场景中如果某个分支任务允许失败但不阻塞下游就需要把trigger_rule改成one_success或者all_done。我曾遇到一个场景每日文件送达事件偶尔有延迟但报表还是要发这时就可以把“等待文件”这个任务设为失败不阻塞。前提是告警必须到位否则失败会被静默吞掉比阻塞更可怕。Airflow在2.x版本里把触发规则设计得更加灵活新的trigger_rulenone_failed也很常用意思是只要有上游没有失败就可以跑适合做“不要求全成功但要求无失败”的聚合任务。3.3 数据在任务之间怎么传XCom与模板变量任务之间要传小数据比如一个文件路径、一个日期参数用XCom。XCom用起来很简单在PythonOperator里return一个值就会自动存入下游可见的上下文里下游用ti.xcom_pull(task_idsxxx)取。但XCom有个坑它把数据存在元数据库里传大DataFrame会把数据库撑爆。大结果的正确做法是写到临时文件或对象存储只把路径传给下游。Airflow的模板变量可以自动渲染字符串参数。{{ ds }}是执行日期{{ data_interval_start }}和{{ data_interval_end }}是当前调度周期的起止时间。把这类模板变量拼进SQL文件路径是数仓定时抽取的常见姿势。注意这些时间受调度时区影响必须先把全局时区配好否则边界会乱。我还习惯把一些全局配置写在DAG文件顶部常量里比如数据源连接ID、目标表名、备份保留天数这样新人接手时不用翻代码找魔法数字。但别把密码、密钥放进DAG文件里要么用Airflow的Connection/Variable存要么用密钥管理服务统一管理。3.4 定时和补数理解execution_dateAirflow的定时语义容易绕晕经常被问“为什么2点的任务execution_date是昨天的日期”。关键点在于每个任务实例表示一个调度周期而不是一个时间点。每天的2点0分0秒是周期的“开始边界”这个周期的实际调度时刻是在下一天的2点也就是说2月3日凌晨2点调度的那次运行它对应的execution_date是2月2日。这在日常使用中没有太大影响一旦做跨天聚合或补数理解这个语义就很重要。比如你需要补跑1月1日到1月31日的数据设置catchupTrue并按天调度Airflow会生成31个任务实例execution_date分别是1月1日到1月31日。这正是它比手工脚本强的地方历史数据出问题改完逻辑后一条命令把一段区间重算一遍。补数时建议用backfill命令行工具而不是直接改start_date。直接改start_date会让DAG从新日期开始重新生成大量历史任务很容易压垮调度器。我踩过一次改了某个DAG的start_date当晚Scheduler直接卡死元数据库连接数飙到极限。4. 生产环境下的真实玩法重试、告警、并发与数仓集成4.1 失败的默认行为不够重试和超时Airflow默认会给DAG设置一些重试参数但为了稳定每个任务还是建议显式配置。retries失败重试次数通常设2到3次。retry_delay两次重试之间的间隔间隔太短容易在同一个故障点反复撞墙。execution_timeout单次运行的最大时长超过则强制失败。这个参数很多人不加结果是任务卡住时整条管道占着worker不放后面的任务全部排队问题表现成“调度越来越慢”。我可以给一个典型配置每天批处理任务重试2次间隔10分钟超时2小时。如果是跟外部API打交道的任务间隔可以拉长甚至用指数退避如果是纯SQL计算超时设短一点快速暴露问题。4.2 告警怎么接邮件和大群机器人任务失败的重试不是终点关键是让人知道。Airflow支持在任务的on_failure_callback里挂回调我一般写一个通用函数把任务信息格式化成一段文本再发给团队群机器人和邮件组。下面是一个简化的回调示例def send_failure_alert(context): dag_id context[dag].dag_id task_id context[task_instance].task_id ts context[ts] log_url context[task_instance].log_url # 这里调用群机器人webhook即可不同平台格式略有差异 # requests.post(webhook_url, json{text: fDAG {dag_id} 任务 {task_id} 失败时间 {ts}日志{log_url}}) return None default_args { owner: data_team, on_failure_callback: send_failure_alert, }回调逻辑要尽量轻别在回调里做重计算或下载大文件因为它是同步执行的太重会影响任务状态更新。另外Scheduler的元数据库里会记录所有事件回调异常不会影响任务运行但会干扰告警接收所以要保证告警通道本身是稳定的。失败告警只是底线我还建议接一个“长任务未完成”的提醒比如某任务超过正常耗时的两倍还没结束就触发提示这能提前发现问题不用等到失败通知。4.3 并发与资源控制Airflow默认并发处理能力比较佛系不加配置时任务多了容易互相挤占。常见参数parallelism整个Airflow实例同时执行的最大任务数默认32CPU核数多可以适当调大。max_active_tasks_per_dag单个DAG最多同时运行的任务数防止某个大DAG把所有worker占没。max_active_runs_per_dagDAG允许同时运行多少个周期比如每小时一次而某个周期跑了很久没这个限制就会堆积。池Pool多DAG争抢资源时的利器比如把“高优任务”和“低优任务”分到不同池低优池再拥挤也不影响高优任务。经验做法是把数据库抽数类的任务放到一个Pool把外部推送任务放到另一个Pool窗口期错开避免同时把数仓连接池打爆。Airflow的UI能看到每个任务等待了多久如果任务一直排队不出多半是池或并发设置太紧而不是代码有问题。4.4 和数仓集成增量抽取与分区表写入Airflow做数仓调度最常用的模式是“写临时表再分区交换”。拿每日增量数据来说上游任务先把当天数据写入staging表跑完检查和抽样后再用一条原子SQL把目标分区数据替换。这样做的好处是失败都在临时表阶段发生目标表始终对外可见不会出现报表查到一半数据的情况。分区表可以按日期参数生成例如log_date2025-01-02作为执行日期SQL里的WHERE event_date {{ ds }}自然用模板变量渲染。流程上用Sensor等上游准备好了再来比如from airflow.sensors.sql import SQLSensor sensor SQLSensor( task_idwait_for_source_table, conn_idsource_dw, sqlSELECT 1 FROM source_schema.source_table WHERE dt {{ ds }} LIMIT 1 )需要注意Sensor默认会周期性轮询间隔设置太短会频繁打数据库默认60秒已经够用不需要更细。检查的分区不是当天而是{{ ds }}代表的历史周期这个要是理解偏了容易出现晚一天才发现没数据。4.5 发布迭代DAG代码的版本与灰度DAG代码也会持续迭代不能今天写完明天直接全量切。我的做法是DAG文件全部进Git仓库禁止直接在服务器上改线上文件。新改动先在测试环境跑通一个周期确认无误再合并到生产分支。关键DAG切换前最好用airflow dags show核对依赖图避免看漏边。如果改动涉及任务ID变更旧的任务实例状态和新任务对不上是正常的注意观察一轮调度周期。Airflow 2.x支持version字段对DAG版本做标注。我会在DAG定义里写一个DAG_VERSION变量管理员从UI就能看到当前跑的是哪个版本排查问题对版本更直接。5. 常见问题排查与避坑实录5.1 任务一直Queued不跑是怎么回事任务状态卡在Queued常见原因有四个并行度打满、池容量被占、Worker数量不够、Scheduler没在正常心跳。排查顺序我固定这样走看UI的DAG运行列表是不是大量任务同时启动挤占了parallelism。看池低优任务是否占满所有槽位高优任务进不来。看Worker的日志最近有没有报错是不是任务进程占满CPU或内存。看Scheduler日志有没有异常或数据库连接报错。这里有个容易忽略的细节如果Scheduler重启过于频繁任务恢复后还会因为状态没写完而卡在Queued。正常情况一个任务从Queued到Running应该在几十秒内超过五分钟还不动优先检查资源而非代码。5.2 时区问题为什么调度时间总是差8小时Airflow默认使用UTC如果不对齐成北京时间UI里所有时间都会比本地晚8小时展示和告警的时间判断都会混乱。配置方法是在airflow.cfg里把时区改为本地时区同时重启Scheduler、Webserver、Worker三件套只改其中一个没用。另外元数据库的连接串也要注意比如PostgreSQL连接时我会配置时区参数避免内部时间字段偏移。任务真正消费的时间参数是以调度周期为准代码里不要再手动加减8小时那样会造成双倍偏移。我第一次做跨天重跑时就是因为手动加8小时补出来的数据日期全部错位排查了大半天才发现。5.3 DAG不显示在UI里代码又没错DAG文件语法没问题但UI里看不到常见原因文件没放到DAG目录Airflow只会扫描dags目录或配置的路径。dag_id和其他DAG重名了UI里只显示一个。Scheduler没有及时发现新文件可以稍等几分钟或者手动触发一次airflow dags reserialize。文件里存在无条件执行的顶层循环或网络请求导致解析超时最终这个DAG文件直接不加载。使用了不存在的自定义Operator或Hook日志里会报导入错误。最烦人的是整体DAG列表空了所有DAG同时消失。出现这种情况先看Scheduler日志多数时候是数据库连接异常或磁盘写满恢复后重新扫描即可。注意airflow.cfg里的dag_dir_list_interval控制扫描间隔调成30秒会响应快一些。5.4 依赖冲突与Python环境混乱Airflow自身的Python依赖非常庞大稍有不慎和项目里其他包冲突整个环境就崩了。生产环境强烈建议把Airflow装在独立虚拟环境或者直接容器化。如果任务里要用很重的数据科学库比如新版NumPy、Pandas尽量把任务执行方式改成独立Python进程或换用KubernetesExecutor让每个任务跑在独立的镜像里。还有一个常见问题多个DAG文件引用了同一个helper模块升级helper后一个DAG正常另一个报错。原因是Scheduler进程里模块缓存未刷新重启Scheduler就能解决。所以生产发版流程里改完公共模块后按步骤重启调度别只重跑单个DAG。5.5 元数据库膨胀与日志管理元数据库记录了所有任务实例、日志、XCom运行几个月后膨胀很快。尤其任务量大的团队几百万行记录会让查询变慢。我定期用Airflow命令行提供的清理方法清理历史日志和任务实例比如清理60天前的任务日志记录、清理大XCom记录。Airflow支持把日志写到远端存储减轻本地磁盘压力同时可以考虑配置日志保留天数按天数自动清理。还有一个容易忽略的点任务日志里的Python堆栈如果层级很深日志文件会非常大任务结束后不清理日积月累磁盘会满。生产上至少留一个磁盘监控磁盘满时Scheduler和数据库都会出连锁问题。最后分享一个我坚持了很久的习惯任何新DAG上线前都会故意让它失败一次确认告警真能送到群里再恢复数据重新跑通一次确保重试逻辑正常。表面看多花了一小时实际上省掉的是后面无数次“悄悄失败”的半夜事故。Airflow的价值不在漂亮的DAG图而在于它把运维里那些容易被人遗忘的细节——依赖、重试、监控、恢复——全部变成了默认能力。把管道交给它之后人才能真正睡个整觉。