Apache Airflow 3 数据工作流调度实践指南：从安装到生产部署

Apache Airflow 3 数据工作流调度实践指南从安装到生产部署【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowApache Airflow 是一个用代码来定义、调度和监控数据工作流的平台属于 Apache 软件基金会顶级项目。它把原本靠人工串联的数据处理步骤写成 Python 代码由调度器按计划自动执行并提供 Web 界面跟踪每一次运行。这篇文章面向第一次接触 Airflow 的开发者带你完成安装、写出第一个工作流并了解日常运营和生产部署的关键点。核心能力全景以代码定义工作流DAGAirflow 中的每个工作流都是一个 DAG有向无环图用 Python 脚本描述任务及依赖关系。代码可以放在版本库里配合 CI/CD 发布工作流变更因此可追溯、可回滚。Airflow 3 还引入了 TaskFlow API直接用task装饰器把普通函数变成任务写起来接近普通 Python 代码。定时调度与触发器定时调度支持 Cron 表达式或daily等预定义间隔到点自动创建新的运行手动触发在 Web 界面或 CLI 中随时手动启动一次运行事件触发Triggerer 组件监听外部事件如文件到达、消息队列信号满足条件时唤醒等待中的任务可观测的执行监控每个任务实例都有完整状态机scheduled → queued → running → success/failedUI 提供 Graph、Grid 两种视图。失败的任务支持配置自动重试运行历史、日志、XCom 数据都可以直接从界面查看排障不用登录服务器翻文件。最小化运行路径环境要求与安装要求 Python 3.10 及以上。最短路径是创建一个虚拟环境安装apache-airflow核心包然后以 standalone 模式启动python3 -m venv airflow-env source airflow-env/bin/activate pip install apache-airflow export AIRFLOW_HOME~/airflow airflow standalonestandalone 模式会一次性拉起调度器、API 服务器、触发器和 Worker 等全部组件适合本地体验和演示。验证点启动成功后终端会打印初始用户名和密码。打开浏览器访问http://localhost:8080用打印的凭据登录看到 DAGs 列表页即代表部署成功。此时列表为空是正常的——DAG 脚本需要放在AIRFLOW_HOME/dags目录下standalone 首次运行会自动创建放入文件后稍等片刻列表就会出现对应工作流。一个端到端场景以每日电商报表为例说明输入→处理→输出的完整链路输入是昨天的订单明细数据处理分三步——清洗汇总、生成指标、推送摘要输出是一份可查的日报和一条推送通知。设计思路是把每步写成独立函数用 TaskFlow API 声明依赖数据质量不达标时后续步骤自动短路。from datetime import datetime, timedelta from airflow import DAG from airflow.decorators import task task def clean_orders(): orders fetch_yesterday_orders() # 读取订单明细 return [o for o in orders if o.valid] task def build_metrics(orders): return {orders: len(orders), gmv: sum(o.amount for o in orders)} task def push_report(metrics): notify(metrics) # 推送到消息渠道 with DAG(dag_iddaily_report, schedule_intervaldaily, start_datedatetime(2025, 1, 1), catchupFalse) as dag: m build_metrics(clean_orders()) push_report(m)任务失败时 Airflow 会按retries配置自动重跑而不需要人工介入整个 DAG 的历史运行状态都能在 UI 的 Runs 视图中回查。更多写法可参考 airflow-core/docs/tutorial/fundamentals.rst。日常运营要点用 Grid 视图快速定位失败任务Grid 视图把任务 × 运行日期展开成矩阵历史失败一目了然点击红色单元格直达该任务实例的日志页侧栏 Overview 面板汇总最近失败的 10 个任务和失败次数趋势对偶发性失败可直接在界面点击重跑无需重发整个 DAG理解任务生命周期再排障每个任务实例的状态流转遵循固定路径参见 airflow-core/docs/img/task_lifecycle_diagram.png 对应的生命周期定义排障时先确认任务卡在哪一环一直 queued检查是否有可用 Worker 并发反复 up_for_retry查看任务日志定位异常确认重试次数配置upstream_failed问题在上游顺藤摸瓜回退检查日志与告警习惯日志统一落在AIRFLOW_HOME/logs界面内可直接查看配置通知邮件、IM 等后失败事件会主动推送避免人工轮询定期审查暂停状态的 DAG长期未使用的及时清理规模化与生产建议容器化部署团队环境建议改用官方 Helm Chart 部署到 KubernetesChart 模板覆盖了调度器、Worker、API 服务器、Postgres、Redis 等全套组件支持按需扩缩容和独立资源配额见 chart/ 目录中的部署文档与参数参考。状态存储与组件分离生产环境把元数据库换成独立的 Postgres/MySQL 实例让调度器与 Worker 分机运行——Airflow 3 的架构中 Worker 通过 Task SDK 与 API 服务器交互用户代码不再直连元数据库这种分离使各组件可以独立横向扩展。快速答疑Q: Airflow 和 crontab 有什么区别A: crontab 只能调度单条命令任务之间没有依赖关系也没有运行状态和历史记录。Airflow 用 DAG 描述任务依赖提供重试、可视化监控和完整的运行历史适合多步骤的数据管道。Q: DAG 文件放哪里修改后多久生效A: 放在AIRFLOW_HOME/dags目录可配置为其他路径。Dag Processor 会轮询该目录文件变化后自动解析并加载新版本无需重启服务。Q: standalone 模式能用于生产吗A: 不建议。standalone 把调度器、API、Worker 全部塞进一个进程适合本地体验和演示。生产环境请使用组件分离的部署方式如官方 Helm Chart。下一步建议先跑通本文的 standalone 环境然后通读 airflow-core/docs/tutorial/ 下的官方教程从 fundamentals 逐步过渡到 TaskFlow 写法。当你把第一个真实的每日任务迁到 Airflow 上并稳定运行一周后再着手评估 Kubernetes 部署路径这样升级时你会对每个组件的职责有更清晰的判断。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考