大数据-253 离线数仓 - Airflow 入门与任务调度实战：DAG、Operator、Executor 部署排错指南

TL;DR场景面向离线数仓与定时任务场景快速理解 Airflow 的核心概念、DAG 编排方式与基础命令。结论本文内容适合作为 Airflow 入门示例但代码与命令明显偏旧需区分 Airflow 1.x 与 2.x 版本差异。产出给出 Airflow 核心概念梳理、版本矩阵、常见报错速查卡便于后续补齐部署与兼容性说明。任务集成部署Airflow 基本介绍Apache Airflow 是一个开源的任务调度和工作流管理工具用于编排复杂的数据处理任务。最初由 Airbnb 开发于 2016 年捐赠给 Apache 软件基金会。Airflow 的主要特点是以代码方式定义任务及其依赖关系支持任务的调度和监控适合处理复杂的大数据任务。Airflow 的特点以代码为中心Airflow 使用 Python 定义 DAG提供灵活性和可编程性。扩展性强用户可以自定义 Operator 和 Hook集成各种数据源和工具。强大的 UI 界面提供可视化界面监控任务状态、查看日志、重试失败任务等。丰富的调度选项支持基于时间 (Time-based) 和事件 (Event-based) 的调度。高可用性配合 Celery 和 Kubernetes 等执行器支持分布式架构适合处理大规模任务。使用场景数据管道调度用于管理数据从源到目标的 ETL 流程。如每天从数据库中抽取数据、清洗后存入数据仓库。机器学习工作流管理调度数据预处理、模型训练和模型部署任务。数据验证自动化检查数据的质量和一致性。定期任务自动化定时清理日志、归档数据或生成报告。Airflow核心概念DAGs有向无环图Directed Acyclic Graph将所有需要运行的tasks按照依赖关系组织起来描述的是所有tasks执行的顺序OperatorsAirflow内置了很多OperatorsBashOperator 执行一个Bash命令PythonOperator 调用任意的Python函数EmailOperator 用于发送邮件HTTPOperator 用于发送HTTP请求SqlOperator 用于执行SQL命令自定义 OperatorTaskTaskTask是Operator的一个实例Task InstanceTask Instance由于Task会被重复调度每次Tasks的运行就是不同的Task InstanceTask Instance 有自己的状态包括 success、running、failed、skipped、up_for_rechedule、up_for_retry、queued、no_status等Task RelationshipsTask RelationshipsDAGs中的不同Tasks之间可以有依赖关系ExecutorExecutor在Airflow中支持的执行器就有四种SequentialExecutor单进程顺序执行任务默认执行器通常只用于测试LocalExecutor多进程本地执行任务CeleryExecutor分布式调度生产常用Celery是一个分布式调度框架其本身没有队列功能需要使用第三方组件如RabbitMQDaskExecutor动态任务调度主要用于数据分析执行器的修改修改 $AIRFLOW_HOME/airflow.cfg 中executor LocalExecutor这里关于执行器的修改修改如下所示入门案例编写脚本mkdir$AIRFLOW_HOME/dagsvim$AIRFLOW_HOME/dags/helloworld.py我们需要写入的内容如下fromdatetimeimportdatetime,timedeltafromairflowimportDAGfromairflow.utilsimportdatesfromairflow.utils.helpersimportchainfromairflow.operators.bash_operatorimportBashOperatorfromairflow.operators.python_operatorimportPythonOperator# 定义默认参数defdefault_options():default_args{owner:airflow,# 拥有者名称start_date:dates.days_ago(1),# 第一次开始执行的时间retries:1,# 失败重试次数retry_delay:timedelta(seconds5)# 失败重试间隔}returndefault_args# 定义Bash任务deftask1(dag):tpwdtaskBashOperator(task_idMyTask1,# task_idbash_commandt,# 指定要执行的命令dagdag# 指定归属的dag)returntask# Python任务函数defhello_world():current_timestr(datetime.today())print(hello world at {}.format(current_time))# 定义Python任务deftask2(dag):taskPythonOperator(task_idMyTask2,python_callablehello_world,# 指定要执行的函数dagdag)returntask# 定义另一个Bash任务deftask3(dag):tdatetaskBashOperator(task_idMyTask3,bash_commandt,dagdag)returntask# 定义DAGwithDAG(HelloWorldDag,# dag_iddefault_argsdefault_options(),# 指定默认参数schedule_interval*/2 * * * *# 执行周期每分钟2次)asd:task1task1(d)task2task2(d)task3task3(d)chain(task1,task2,task3)# 指定执行顺序写入的内容如下所示测试运行# 执行命令检查脚本是否有错误。如果命令行没有报错就表示没问题python$AIRFLOW_HOME/dags/helloworld.py执行的结果如下图所示查看生效的 dags# 查看生效的 dagsairflow dags list--subdir$AIRFLOW_HOME/dags执行结果如下图所示查看指定dag中的taskairflow tasks list HelloWorldDag执行的结果如下图所示测试dag中的taskairflow taskstestHelloWorldDag MyTask22020-08-01执行的结果如下所示错误速查症状根因定位修复ModuleNotFoundError: airflow.operators.bash_operator使用了 Airflow 1.x 导包方式但当前环境是 Airflow 2.x执行 DAG 文件或 Scheduler 加载日志改为 2.x 对应导包路径并在文中标注适用版本DAG 文件存在但 airflow dags list 看不到DAG 未被成功解析代码异常文件路径不在 dags_folder查看 Scheduler / Webserver 日志执行 airflow dags list --subdir先修复导包或语法错误再确认 $AIRFLOW_HOME/dags 配置airflow tasks test 报 DAG 或 Task 不存在dag_id / task_id 写错或 DAG 未注册成功用 airflow dags list、airflow tasks list HelloWorldDag 核对保证 dag_idHelloWorldDag、task_idMyTask2 与脚本一致DAG 能加载但任务不按预期调度start_date、调度周期、时区配置不清晰查看 DAG 详情页中的 next run / logical date显式写清 start_date 和时区避免只用 days_ago()文章说“每分钟 2 次”但 cron 实际含义有歧义*/2 * * * * 实际是每 2 分钟一次对照 cron 表达式与 Airflow UI 运行时间将文案改为“每 2 分钟执行一次”修改 executor LocalExecutor 后任务仍异常只改配置未满足依赖条件如数据库未切换到 MySQL/PostgreSQL查看 airflow.cfg、元数据库类型、启动日志补齐 LocalExecutor/CeleryExecutor 的数据库与组件要求CeleryExecutor 启动失败Broker / Result Backend 未配置完整查看 Celery Worker / Scheduler 日志明确 Redis 或 RabbitMQ 配置不要只写 RabbitMQ直接执行 python dags/helloworld.py 没报错但 UI 仍无 DAG语法没问题但 Airflow 上下文加载失败对比命令行结果与 Scheduler 解析日志以 Airflow 自身的 DAG 解析结果为准不把 python 执行当最终验证任务状态里出现 up_for_reschedule / up_for_retry 表述混乱状态名或拼写不准确版本差异导致状态集合不同查任务实例详情页或官方状态枚举避免在正文里罗列过多状态名改成“常见状态如 success、running、failed、retry”执行器说明被读者质疑不完整对执行器支持范围下了绝对结论对照当前 Airflow 版本官方文档改成“常见执行器包括……”而不是“只有四种”其他系列 AI篇持续更新中长期更新AI炼丹日志-29 - 字节跳动 DeerFlow 深度研究框斜体样式架 私有部署 测试上手 架构研究持续打造实用AI工具指南AI研究-132 Java 生态前沿 2025Spring、Quarkus、GraalVM、CRaC 与云原生落地 AI模块直达链接 Java篇持续更新中长期更新Java-218 RocketMQ Java API 实战同步/异步 Producer 与 Pull/Push ConsumerMyBatis 已完结Spring 已完结Nginx已完结Tomcat已完结分布式服务已完结Dubbo已完结MySQL已完结MongoDB已完结Neo4j已完结FastDFS 已完结OSS已完结GuavaCache已完结EVCache已完结RabbitMQ已完结RocketMQ正在更新… 深入浅出助你打牢基础 Java模块直达链接 大数据板块已完成多项干货更新300篇包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件覆盖离线实时数仓全栈大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT案例 详解 大数据模块直达链接