1. 从“胶水脚本”到“调度工厂”为什么数据工程师需要Airflow如果你是一名数据工程师或者正在处理任何与数据流水线相关的工作那么你一定经历过这样的场景凌晨三点被手机警报吵醒原因是某个关键的ETL任务失败了。你睡眼惺忪地爬起来登录服务器在一堆杂乱的日志文件里寻找线索发现是因为上游数据源格式变了或者某个依赖的API服务挂了又或者仅仅是服务器内存不足。你手动修复、重跑祈祷下一次能正常运行。这种日子我们称之为“脚本小子”的黑暗时代——用Python写一堆独立的脚本再用crontab定时触发脚本之间靠文件或数据库状态来隐式通信监控和错误处理全靠人工盯梢和“printf大法”。这种模式在小规模、简单的场景下或许还能应付但随着数据源增多、处理逻辑复杂、任务依赖关系像蜘蛛网一样交织时它就彻底崩溃了。你需要的不是一个更强大的crontab而是一个工作流编排与调度系统。这就是Apache Airflow诞生的背景也是它成为数据工程师“标配工具”的根本原因。Airflow的核心思想是用代码Python来定义、调度和监控工作流。它把那些散落的、脆弱的脚本变成了一个可视化、可监控、可回溯、具备强大依赖管理和错误恢复能力的“调度工厂”。简单来说Airflow让你告别“胶水胶带”式的运维进入“声明式编排”的工业时代。你不再需要关心“什么时候该运行哪个脚本”而是专注于定义“任务之间应该是什么关系”。剩下的比如定时触发、依赖检查、任务执行、失败重试、日志收集、状态监控全部交给Airflow。这听起来可能有点抽象但当你真正用它构建起第一条从数据抽取、清洗、转换到加载入库的完整流水线并看着它在Web界面上清晰流转时你会立刻明白它的价值。2. Airflow核心架构拆解DAG、Operator与Executor是如何协同工作的要理解Airflow必须吃透它的三个核心概念DAG、Operator和Executor。这三者构成了Airflow调度引擎的骨架。2.1 DAG工作流的蓝图DAG全称有向无环图是Airflow中最核心的抽象。你可以把它理解为一个项目的任务流程图。这个图由多个任务节点组成节点之间有明确的依赖关系箭头指向并且不能形成循环。一个DAG定义了一个完整的工作流。在代码中一个DAG就是一个Python对象。它最重要的属性是dag_id唯一标识和schedule_interval调度间隔比如daily或cron表达式。DAG文件通常存放在Airflow的DAGS_FOLDER目录下Airflow的调度器会定期扫描这个文件夹解析其中的DAG定义。from airflow import DAG from datetime import datetime, timedelta default_args { owner: data_team, depends_on_past: False, email_on_failure: True, email: [alertexample.com], retries: 3, retry_delay: timedelta(minutes5), } # 实例化一个DAG对象 with DAG( dag_idmy_etl_pipeline, # DAG的唯一ID default_argsdefault_args, description一个简单的ETL示例流水线, schedule_interval0 2 * * *, # 每天凌晨2点运行 start_datedatetime(2023, 10, 1), catchupFalse, # 是否补跑历史任务 tags[example, etl], ) as dag: # 在这里定义任务Operators pass关键理解DAG本身不执行任何操作它只是一个蓝图定义了“谁”任务在“什么时候”调度以“什么顺序”依赖执行。start_date和schedule_interval共同决定了DAG RunDAG的一次执行实例的产生时间。catchup参数至关重要如果设为TrueAirflow会从start_date开始为每一个过去的调度周期都创建一个DAG Run这可能导致“任务海啸”在生产中需谨慎开启。2.2 Operator任务的具体执行者如果说DAG是蓝图那么Operator就是蓝图上的一个个具体工种。每个Operator代表一个独立的执行单元。Airflow提供了丰富的内置Operator比如BashOperator: 执行一个bash命令。PythonOperator: 执行一个Python函数。EmailOperator: 发送邮件。SimpleHttpOperator: 发送HTTP请求。DockerOperator: 在Docker容器中运行任务。KubernetesPodOperator: 在K8s Pod中运行任务。当你说“我需要运行一个Spark作业”时你可能会使用SparkSubmitOperator当你说“需要检查HDFS上某个文件是否存在”时你会用Sensor传感器一种特殊的Operator。任务之间的依赖关系通过位运算符下游和上游来设置非常直观。from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator def process_data(**context): # 你的数据处理逻辑 data context[ti].xcom_pull(task_idsextract_task) # 从上游任务获取数据 print(fProcessing: {data}) return processed_data # 定义任务 extract_task BashOperator( task_idextract_task, bash_commandecho raw_data, dagdag, ) process_task PythonOperator( task_idprocess_task, python_callableprocess_data, dagdag, ) load_task BashOperator( task_idload_task, bash_commandecho Loading to database..., dagdag, ) # 定义依赖extract - process - load extract_task process_task load_task实操心得选择Operator时一个重要的原则是“幂等性”。即任务无论执行多少次只要输入相同结果都应该相同。这保证了任务失败重试时不会产生副作用。对于不符合幂等性的操作比如向一个表追加数据需要额外设计例如先清空目标表再全量插入或者使用更精细的增量逻辑。2.3 Executor任务的执行引擎Executor决定了任务在哪里以及如何被执行。它是Airflow可扩展性的关键。常见的Executor有SequentialExecutor: 顺序执行器一次只执行一个任务。仅用于开发和测试因为它使用SQLite数据库无法并行。LocalExecutor: 本地执行器在调度器所在的机器上使用多进程并行执行任务。适用于中小规模部署。CeleryExecutor: 分布式执行器使用Celery作为分布式任务队列可以将任务分发到多台工作节点上执行。这是生产环境最常用的模式提供了良好的水平扩展能力。KubernetesExecutor: 每个任务都会动态地在Kubernetes集群中启动一个独立的Pod来执行。任务彼此隔离资源利用率高但复杂度也更高。配置选择背后的逻辑如果你的团队已经有Kubernetes集群并且希望实现极致的资源隔离和弹性KubernetesExecutor是很好的选择。但对于大多数数据平台团队CeleryExecutor搭配一批专用的工作节点Worker是更成熟、更易于运维的方案。它平衡了复杂度、功能和社区支持。Airflow的调度器Scheduler作为一个常驻进程负责解析DAG、根据调度周期创建DAG Run、检查任务依赖并将满足执行条件的任务实例Task Instance放入执行队列。Worker进程当使用CeleryExecutor时则从队列中取出任务并执行。Web Server提供UI界面用于监控和管理。元数据数据库如PostgreSQL/MySQL存储所有DAG、任务、变量、连接等信息。3. 从零到一手把手搭建一个生产可用的Airflow环境纸上得来终觉浅绝知此事要躬行。下面我们以最常用的CeleryExecutor模式为例搭建一个可用于生产原型的环境。我们假设使用Linux系统并以PostgreSQL作为元数据库Redis作为Celery的消息队列。3.1 基础环境与依赖安装首先确保系统已安装Python建议3.8和pip。强烈建议使用虚拟环境如venv或conda来隔离Airflow的依赖。# 创建并激活虚拟环境 python -m venv airflow_venv source airflow_venv/bin/activate # 安装Airflow。生产环境通常固定版本这里安装2.7.1版本并指定PostgreSQL和Celery支持 pip install apache-airflow[celery, postgres, redis, password]2.7.1 --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.7.1/constraints-3.8.txt注意--constraint参数至关重要它指定了与该版本Airflow兼容的依赖包版本能极大避免因依赖冲突导致的环境问题。请根据你的Python版本修改URL中的数字。3.2 数据库与消息队列配置安装并启动PostgreSQL和Redis如果尚未安装# Ubuntu/Debian 示例 sudo apt-get update sudo apt-get install -y postgresql postgresql-contrib redis-server sudo systemctl start postgresql sudo systemctl start redis创建Airflow数据库和用户sudo -u postgres psql在PostgreSQL命令行中执行CREATE DATABASE airflow_db; CREATE USER airflow_user WITH PASSWORD your_secure_password; GRANT ALL PRIVILEGES ON DATABASE airflow_db TO airflow_user; \q配置AirflowAirflow的配置文件位于~/airflow/airflow.cfg首次运行airflow db init后生成。我们需要修改几个关键配置[core] # 设置Executor类型 executor CeleryExecutor # 设置数据库连接替换为你的实际密码和主机 sql_alchemy_conn postgresqlpsycopg2://airflow_user:your_secure_passwordlocalhost/airflow_db # 设置DAG文件夹路径 dags_folder /path/to/your/dags # 关闭示例DAG生产环境建议关闭 load_examples False [celery] # 设置Celery Broker URL (Redis) broker_url redis://localhost:6379/0 # 设置Celery Result Backend (通常也用数据库) result_backend dbpostgresql://airflow_user:your_secure_passwordlocalhost/airflow_db3.3 初始化与启动服务初始化数据库这会在元数据库中创建所有必要的表。airflow db init创建管理员用户用于登录Web UI。airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com \ --password admin123启动服务生产环境通常使用systemd或supervisor来管理进程。这里我们先在终端启动以测试。启动Web Server守护进程airflow webserver --port 8080 --daemon启动Scheduler守护进程airflow scheduler --daemon启动Celery Worker指定队列名例如defaultairflow celery worker --daemon现在访问http://localhost:8080用刚才创建的用户名密码登录你应该能看到Airflow的Web UI了。如果看不到DAG请检查你的dags_folder路径下是否有Python DAG文件。踩坑实录权限与路径问题初次部署最常见的两个问题是1) PostgreSQL的pg_hba.conf配置导致本地连接认证失败需要确保local和host记录配置正确。2) Worker执行任务时因为运行用户不同可能没有权限读取DAG文件或执行脚本。确保Airflow的各个组件Web Server, Scheduler, Worker运行用户对相关路径有读取和执行权限。一个简单的测试方法是切换到运行Airflow服务的用户手动执行一下DAG中的Bash命令或Python脚本。4. 编写你的第一个生产级DAG一个真实的ETL案例让我们构建一个稍微复杂一点的、贴近生产场景的DAG。假设我们每天需要从某个API获取用户行为日志清洗后与数据库中的用户维度表关联最后将结果写入数据仓库如ClickHouse并发送汇总报告邮件。4.1 项目结构与DAG设计一个好的实践是为每个DAG或相关DAG组建立一个独立的文件夹并利用Python的模块化能力。my_data_pipelines/ ├── dags/ │ └── user_behavior_etl.py # 主DAG文件 ├── plugins/ # 自定义Operator或Hook ├── scripts/ # 独立的SQL或Shell脚本 ├── sql/ │ └── transform_user_behavior.sql └── config/ └── constants.py # 存放配置常量DAG代码实现(user_behavior_etl.py)from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator, BranchPythonOperator from airflow.operators.bash import BashOperator from airflow.operators.email import EmailOperator from airflow.operators.dummy import DummyOperator from airflow.providers.http.sensors.http import HttpSensor from airflow.providers.postgres.operators.postgres import PostgresOperator from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator # 假设我们使用社区提供的ClickHouse Provider from airflow.providers.clickhouse.operators.clickhouse import ClickHouseOperator from airflow.utils.edgemodifier import Label import sys sys.path.append(/path/to/my_data_pipelines) from config.constants import API_ENDPOINT, EMAIL_LIST default_args { owner: bi_team, depends_on_past: False, email_on_failure: True, email_on_retry: False, retries: 3, retry_delay: timedelta(minutes2), execution_timeout: timedelta(hours2), } def _extract_api_data(**context): 从API提取数据并推送到XCom供下游任务使用 import requests import pandas as pd execution_date context[execution_date] # 构造请求参数例如获取前一天的数据 params {date: (execution_date - timedelta(days1)).strftime(%Y-%m-%d)} response requests.get(API_ENDPOINT, paramsparams) response.raise_for_status() data response.json() # 简单转换这里可以更复杂 df pd.DataFrame(data[records]) # 将数据以JSON字符串形式推送到XCom context[ti].xcom_push(keyapi_raw_data, valuedf.to_json(orientrecords)) return fExtracted {len(df)} records. def _process_data(**context): 处理数据清洗、去重、转换 import pandas as pd import json ti context[ti] raw_data_json ti.xcom_pull(task_idsextract_from_api, keyapi_raw_data) df pd.read_json(raw_data_json, orientrecords) # 执行清洗逻辑例如去重、处理空值、类型转换 df_cleaned df.drop_duplicates(subset[user_id, event_time]) df_cleaned[event_time] pd.to_datetime(df_cleaned[event_time]) # 再次推送到XCom或者也可以直接写入临时存储如S3/MinIO ti.xcom_push(keycleaned_data, valuedf_cleaned.to_json(orientrecords)) return fProcessed {len(df_cleaned)} records. def _check_data_quality(**context): 数据质量检查决定是否继续执行加载任务 import pandas as pd import json ti context[ti] cleaned_data_json ti.xcom_pull(task_idsprocess_and_clean, keycleaned_data) df pd.read_json(cleaned_data_json, orientrecords) # 简单的质量检查规则 if df.empty: return alert_empty_data # 返回下游任务的task_id elif df[user_id].isnull().sum() len(df) * 0.1: # 超过10%的user_id为空 return alert_bad_quality else: return load_to_warehouse with DAG( dag_iduser_behavior_daily_etl, default_argsdefault_args, description每日用户行为日志ETL流水线, schedule_interval30 1 * * *, # 每天凌晨1:30运行 start_datedatetime(2024, 1, 1), catchupFalse, max_active_runs1, # 防止并发导致的数据混乱 tags[production, etl, user_behavior], ) as dag: start DummyOperator(task_idstart) end DummyOperator(task_idend) # 1. 检查API是否可用传感器任务 api_available HttpSensor( task_idapi_available, http_conn_iduser_behavior_api, # 需要在Airflow UI中预先配置Connection endpoint/health, response_checklambda response: response.status_code 200, poke_interval30, timeout5*60, modepoke, ) # 2. 提取数据 extract_from_api PythonOperator( task_idextract_from_api, python_callable_extract_api_data, ) # 3. 清洗处理 process_and_clean PythonOperator( task_idprocess_and_clean, python_callable_process_data, ) # 4. 数据质量检查与分支 data_quality_check BranchPythonOperator( task_iddata_quality_check, python_callable_check_data_quality, ) # 5. 分支路径正常加载 load_to_warehouse ClickHouseOperator( task_idload_to_warehouse, clickhouse_conn_idclickhouse_prod, sql INSERT INTO dw.user_behavior_daily SELECT user_id, event_type, event_time, device_id, ... FROM input(user_id String, event_type String, ...) FORMAT JSONEachRow , parameters{data: {{ ti.xcom_pull(task_idsprocess_and_clean, keycleaned_data) }}}, ) # 6. 分支路径告警 alert_empty_data EmailOperator( task_idalert_empty_data, toEMAIL_LIST, subjectAirflow Alert: 用户行为数据为空 - {{ ds }}, html_contentp今日用户行为API返回数据为空请检查数据源。/p, ) alert_bad_quality EmailOperator( task_idalert_bad_quality, toEMAIL_LIST, subjectAirflow Alert: 用户行为数据质量异常 - {{ ds }}, html_contentp今日用户行为数据质量检查未通过如大量空值请核查。/p, ) # 7. 后续步骤数据聚合仅当成功加载后执行 run_daily_aggregation PostgresOperator( task_idrun_daily_aggregation, postgres_conn_idpostgres_warehouse, sqlsql/transform_user_behavior.sql, # 引用外部SQL文件 ) send_daily_report EmailOperator( task_idsend_daily_report, toEMAIL_LIST, subject每日用户行为报告 - {{ ds }}, html_contentpETL任务已完成详细报告见附件或BI系统。/p, ) # 定义依赖关系 start api_available extract_from_api process_and_clean data_quality_check # 分支依赖 data_quality_check [load_to_warehouse, alert_empty_data, alert_bad_quality] # 主路径继续 load_to_warehouse run_daily_aggregation send_daily_report end # 告警路径直接结束 alert_empty_data end alert_bad_quality end4.2 代码详解与避坑指南使用XCom进行小数据量传递XCom是Airflow中任务间传递数据的机制。但请注意XCom不适合传递大型数据集如几百MB的数据因为数据会存储在元数据库中。对于大数据最佳实践是将中间数据写入共享存储如S3、HDFS、NFS然后传递文件路径。上述示例仅用于演示小数据场景。连接管理代码中的http_conn_id、postgres_conn_id等需要在Airflow Web UI的Admin - Connections中预先配置。这是集中管理凭证和连接信息的安全方式避免将密码硬编码在DAG中。模板变量Airflow支持Jinja2模板。{{ ds }}是内置变量代表当前执行日期的“YYYY-MM-DD”格式字符串。在EmailOperator的标题中使用它可以让告警邮件自带日期上下文非常实用。BranchPythonOperator这是一个强大的操作符允许你根据上游任务的输出动态决定下游执行路径。它返回的是下一个要执行的task_id。这实现了工作流中的条件逻辑。执行超时与重试在default_args中设置了execution_timeout和retries。对于长时间运行的任务如Spark作业需要合理设置超时时间。重试机制配合retry_delay可以应对短暂的网络抖动或资源不足。依赖关系清晰化使用和运算符让依赖关系一目了然。对于复杂的依赖可以使用set_upstream和set_downstream方法但位运算符更简洁。一个真实的坑时区问题。Airflow默认使用UTC时间。如果你的业务系统使用本地时间如东八区在定义schedule_interval和start_date时以及在任务逻辑中处理execution_date时必须非常小心。建议在DAG级别或全局配置中统一时区并在所有时间处理逻辑中显式进行时区转换。5. 进阶实战性能调优、监控与运维最佳实践当你的Airflow集群开始承载成百上千个DAG和任务时性能和运维就成了挑战。以下是一些来自实战的经验。5.1 性能调优关键点调度器性能[scheduler] parsing_processes增加DAG解析进程数加快DAG文件解析速度。通常设置为CPU核心数。[scheduler] max_threads增加调度器主线程数用于处理任务调度逻辑。减少DAG文件的复杂度避免在DAG文件的顶层模块中执行耗时的操作如大数据量查询、网络请求。将这些逻辑移到Operator内部或辅助函数中。调度器会频繁导入DAG文件复杂的顶层代码会成为瓶颈。使用.airflowignore文件忽略不需要被扫描的目录或文件减少调度器负担。Executor与WorkerCelery Worker并发度通过-c参数或配置worker_concurrency调整每个Worker可以同时执行的任务数。不要超过Worker机器的CPU核心数。任务资源隔离对于资源消耗大CPU/内存的任务使用KubernetesPodOperator或DockerOperator进行隔离避免影响其他任务。也可以使用Celery的任务队列将重型任务路由到专用的高配Worker节点。优化任务粒度避免设计运行时间过长的“巨无霸”任务。将其拆分为多个小的、可重试的原子任务。这提高了并行度和失败恢复的灵活性。数据库优化Airflow的元数据库在运行中会产生大量记录TaskInstance, Log, XCom等。必须定期清理。使用Airflow内置的airflow db clean命令2.0版本后或维护DAG来删除旧数据。否则数据库会无限膨胀导致性能急剧下降。对于PostgreSQL考虑对task_instance、log等大表建立合适的索引如dag_id,state,execution_date。5.2 监控与告警体系“没有监控的系统就是在裸奔。” Airflow的监控分为几个层次内置UI监控Tree View、Graph View、Gantt Chart是基本的任务状态和运行时长可视化工具。重点关注长时间运行、失败或重试状态的任务。日志管理Airflow默认将任务日志存储在本地文件系统。生产环境应集成到集中式日志系统如ELK Stack、LokiGraylog。配置[logging] remote_logging相关设置将日志发送到远程存储如S3、GCS并供日志系统抓取。这样可以在一个地方查看所有任务的日志方便排错。指标暴露Airflow内置了StatsD指标可以集成到PrometheusGrafana中。关键指标包括scheduler_heartbeat调度器是否存活。executor.open_slots/executor.queued_tasksExecutor资源使用情况。dagrun.duration.*/dagrun.schedule_delay.*DAG运行时长和调度延迟。task_instance.*任务状态计数成功、失败、重试。 在Grafana中绘制这些指标的仪表盘可以实时掌握集群健康度。主动告警任务失败告警利用email_on_failure参数是最基本的。更高级的做法是使用on_failure_callback在任务失败时触发一个自定义的Python函数可以集成到企业微信、钉钉、Slack、PagerDuty等告警平台。DAG运行超时告警使用on_retry_callback或专门的监控DAG来检查其他DAG的运行状态。可以编写一个传感器DAG定期检查关键DAG的最新运行状态是否成功、是否按时完成。Scheduler/Worker进程存活告警通过监控系统如Prometheus的up指标或简单的HTTP探针监控Airflow各个服务的进程。5.3 版本控制与CI/CD将DAG代码视为应用程序代码纳入Git版本控制。并建立CI/CD流水线代码检查在CI阶段运行pylint、black、mypy等工具确保代码风格和质量。单元测试使用Airflow的测试工具如airflow.utils.dag_cycle_tester检查循环依赖和pytest对自定义的Python函数、Operator进行单元测试。集成测试可以搭建一个与生产环境隔离的测试Airflow环境将DAG部署上去触发一次运行验证整个流程是否通畅。自动部署CD阶段将通过测试的DAG代码同步到生产环境的DAGS_FOLDER目录。可以使用rsync、K8s ConfigMap或专门的部署工具。一个至关重要的经验DAG的幂等性与数据日期。在设计DAG时必须确保它能够处理“重跑”场景。这意味着任务逻辑应该基于execution_date或data_interval_start/end来确定它要处理的数据范围而不是“今天”或“现在”。这样当你需要重新处理某一天的历史数据时只需要清除那一天的DAG Run状态并重新触发即可任务会自动处理正确的数据切片而不会影响到其他日期的数据。这是构建健壮数据流水线的基石。Airflow不是一个“安装即忘”的工具它是一个需要精心设计和持续运维的数据编排平台。从简单的任务调度起步逐步深入到依赖管理、错误处理、性能优化和监控告警你会逐渐构建起一个可靠、高效、可维护的数据基础设施。这个过程充满挑战但当你看到复杂的数据流在Airflow的调度下井然有序地自动运行时那种掌控感和效率提升会让所有的投入都变得值得。
1. 从“胶水脚本”到“调度工厂”为什么数据工程师需要Airflow如果你是一名数据工程师或者正在处理任何与数据流水线相关的工作那么你一定经历过这样的场景凌晨三点被手机警报吵醒原因是某个关键的ETL任务失败了。你睡眼惺忪地爬起来登录服务器在一堆杂乱的日志文件里寻找线索发现是因为上游数据源格式变了或者某个依赖的API服务挂了又或者仅仅是服务器内存不足。你手动修复、重跑祈祷下一次能正常运行。这种日子我们称之为“脚本小子”的黑暗时代——用Python写一堆独立的脚本再用crontab定时触发脚本之间靠文件或数据库状态来隐式通信监控和错误处理全靠人工盯梢和“printf大法”。这种模式在小规模、简单的场景下或许还能应付但随着数据源增多、处理逻辑复杂、任务依赖关系像蜘蛛网一样交织时它就彻底崩溃了。你需要的不是一个更强大的crontab而是一个工作流编排与调度系统。这就是Apache Airflow诞生的背景也是它成为数据工程师“标配工具”的根本原因。Airflow的核心思想是用代码Python来定义、调度和监控工作流。它把那些散落的、脆弱的脚本变成了一个可视化、可监控、可回溯、具备强大依赖管理和错误恢复能力的“调度工厂”。简单来说Airflow让你告别“胶水胶带”式的运维进入“声明式编排”的工业时代。你不再需要关心“什么时候该运行哪个脚本”而是专注于定义“任务之间应该是什么关系”。剩下的比如定时触发、依赖检查、任务执行、失败重试、日志收集、状态监控全部交给Airflow。这听起来可能有点抽象但当你真正用它构建起第一条从数据抽取、清洗、转换到加载入库的完整流水线并看着它在Web界面上清晰流转时你会立刻明白它的价值。2. Airflow核心架构拆解DAG、Operator与Executor是如何协同工作的要理解Airflow必须吃透它的三个核心概念DAG、Operator和Executor。这三者构成了Airflow调度引擎的骨架。2.1 DAG工作流的蓝图DAG全称有向无环图是Airflow中最核心的抽象。你可以把它理解为一个项目的任务流程图。这个图由多个任务节点组成节点之间有明确的依赖关系箭头指向并且不能形成循环。一个DAG定义了一个完整的工作流。在代码中一个DAG就是一个Python对象。它最重要的属性是dag_id唯一标识和schedule_interval调度间隔比如daily或cron表达式。DAG文件通常存放在Airflow的DAGS_FOLDER目录下Airflow的调度器会定期扫描这个文件夹解析其中的DAG定义。from airflow import DAG from datetime import datetime, timedelta default_args { owner: data_team, depends_on_past: False, email_on_failure: True, email: [alertexample.com], retries: 3, retry_delay: timedelta(minutes5), } # 实例化一个DAG对象 with DAG( dag_idmy_etl_pipeline, # DAG的唯一ID default_argsdefault_args, description一个简单的ETL示例流水线, schedule_interval0 2 * * *, # 每天凌晨2点运行 start_datedatetime(2023, 10, 1), catchupFalse, # 是否补跑历史任务 tags[example, etl], ) as dag: # 在这里定义任务Operators pass关键理解DAG本身不执行任何操作它只是一个蓝图定义了“谁”任务在“什么时候”调度以“什么顺序”依赖执行。start_date和schedule_interval共同决定了DAG RunDAG的一次执行实例的产生时间。catchup参数至关重要如果设为TrueAirflow会从start_date开始为每一个过去的调度周期都创建一个DAG Run这可能导致“任务海啸”在生产中需谨慎开启。2.2 Operator任务的具体执行者如果说DAG是蓝图那么Operator就是蓝图上的一个个具体工种。每个Operator代表一个独立的执行单元。Airflow提供了丰富的内置Operator比如BashOperator: 执行一个bash命令。PythonOperator: 执行一个Python函数。EmailOperator: 发送邮件。SimpleHttpOperator: 发送HTTP请求。DockerOperator: 在Docker容器中运行任务。KubernetesPodOperator: 在K8s Pod中运行任务。当你说“我需要运行一个Spark作业”时你可能会使用SparkSubmitOperator当你说“需要检查HDFS上某个文件是否存在”时你会用Sensor传感器一种特殊的Operator。任务之间的依赖关系通过位运算符下游和上游来设置非常直观。from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator def process_data(**context): # 你的数据处理逻辑 data context[ti].xcom_pull(task_idsextract_task) # 从上游任务获取数据 print(fProcessing: {data}) return processed_data # 定义任务 extract_task BashOperator( task_idextract_task, bash_commandecho raw_data, dagdag, ) process_task PythonOperator( task_idprocess_task, python_callableprocess_data, dagdag, ) load_task BashOperator( task_idload_task, bash_commandecho Loading to database..., dagdag, ) # 定义依赖extract - process - load extract_task process_task load_task实操心得选择Operator时一个重要的原则是“幂等性”。即任务无论执行多少次只要输入相同结果都应该相同。这保证了任务失败重试时不会产生副作用。对于不符合幂等性的操作比如向一个表追加数据需要额外设计例如先清空目标表再全量插入或者使用更精细的增量逻辑。2.3 Executor任务的执行引擎Executor决定了任务在哪里以及如何被执行。它是Airflow可扩展性的关键。常见的Executor有SequentialExecutor: 顺序执行器一次只执行一个任务。仅用于开发和测试因为它使用SQLite数据库无法并行。LocalExecutor: 本地执行器在调度器所在的机器上使用多进程并行执行任务。适用于中小规模部署。CeleryExecutor: 分布式执行器使用Celery作为分布式任务队列可以将任务分发到多台工作节点上执行。这是生产环境最常用的模式提供了良好的水平扩展能力。KubernetesExecutor: 每个任务都会动态地在Kubernetes集群中启动一个独立的Pod来执行。任务彼此隔离资源利用率高但复杂度也更高。配置选择背后的逻辑如果你的团队已经有Kubernetes集群并且希望实现极致的资源隔离和弹性KubernetesExecutor是很好的选择。但对于大多数数据平台团队CeleryExecutor搭配一批专用的工作节点Worker是更成熟、更易于运维的方案。它平衡了复杂度、功能和社区支持。Airflow的调度器Scheduler作为一个常驻进程负责解析DAG、根据调度周期创建DAG Run、检查任务依赖并将满足执行条件的任务实例Task Instance放入执行队列。Worker进程当使用CeleryExecutor时则从队列中取出任务并执行。Web Server提供UI界面用于监控和管理。元数据数据库如PostgreSQL/MySQL存储所有DAG、任务、变量、连接等信息。3. 从零到一手把手搭建一个生产可用的Airflow环境纸上得来终觉浅绝知此事要躬行。下面我们以最常用的CeleryExecutor模式为例搭建一个可用于生产原型的环境。我们假设使用Linux系统并以PostgreSQL作为元数据库Redis作为Celery的消息队列。3.1 基础环境与依赖安装首先确保系统已安装Python建议3.8和pip。强烈建议使用虚拟环境如venv或conda来隔离Airflow的依赖。# 创建并激活虚拟环境 python -m venv airflow_venv source airflow_venv/bin/activate # 安装Airflow。生产环境通常固定版本这里安装2.7.1版本并指定PostgreSQL和Celery支持 pip install apache-airflow[celery, postgres, redis, password]2.7.1 --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.7.1/constraints-3.8.txt注意--constraint参数至关重要它指定了与该版本Airflow兼容的依赖包版本能极大避免因依赖冲突导致的环境问题。请根据你的Python版本修改URL中的数字。3.2 数据库与消息队列配置安装并启动PostgreSQL和Redis如果尚未安装# Ubuntu/Debian 示例 sudo apt-get update sudo apt-get install -y postgresql postgresql-contrib redis-server sudo systemctl start postgresql sudo systemctl start redis创建Airflow数据库和用户sudo -u postgres psql在PostgreSQL命令行中执行CREATE DATABASE airflow_db; CREATE USER airflow_user WITH PASSWORD your_secure_password; GRANT ALL PRIVILEGES ON DATABASE airflow_db TO airflow_user; \q配置AirflowAirflow的配置文件位于~/airflow/airflow.cfg首次运行airflow db init后生成。我们需要修改几个关键配置[core] # 设置Executor类型 executor CeleryExecutor # 设置数据库连接替换为你的实际密码和主机 sql_alchemy_conn postgresqlpsycopg2://airflow_user:your_secure_passwordlocalhost/airflow_db # 设置DAG文件夹路径 dags_folder /path/to/your/dags # 关闭示例DAG生产环境建议关闭 load_examples False [celery] # 设置Celery Broker URL (Redis) broker_url redis://localhost:6379/0 # 设置Celery Result Backend (通常也用数据库) result_backend dbpostgresql://airflow_user:your_secure_passwordlocalhost/airflow_db3.3 初始化与启动服务初始化数据库这会在元数据库中创建所有必要的表。airflow db init创建管理员用户用于登录Web UI。airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com \ --password admin123启动服务生产环境通常使用systemd或supervisor来管理进程。这里我们先在终端启动以测试。启动Web Server守护进程airflow webserver --port 8080 --daemon启动Scheduler守护进程airflow scheduler --daemon启动Celery Worker指定队列名例如defaultairflow celery worker --daemon现在访问http://localhost:8080用刚才创建的用户名密码登录你应该能看到Airflow的Web UI了。如果看不到DAG请检查你的dags_folder路径下是否有Python DAG文件。踩坑实录权限与路径问题初次部署最常见的两个问题是1) PostgreSQL的pg_hba.conf配置导致本地连接认证失败需要确保local和host记录配置正确。2) Worker执行任务时因为运行用户不同可能没有权限读取DAG文件或执行脚本。确保Airflow的各个组件Web Server, Scheduler, Worker运行用户对相关路径有读取和执行权限。一个简单的测试方法是切换到运行Airflow服务的用户手动执行一下DAG中的Bash命令或Python脚本。4. 编写你的第一个生产级DAG一个真实的ETL案例让我们构建一个稍微复杂一点的、贴近生产场景的DAG。假设我们每天需要从某个API获取用户行为日志清洗后与数据库中的用户维度表关联最后将结果写入数据仓库如ClickHouse并发送汇总报告邮件。4.1 项目结构与DAG设计一个好的实践是为每个DAG或相关DAG组建立一个独立的文件夹并利用Python的模块化能力。my_data_pipelines/ ├── dags/ │ └── user_behavior_etl.py # 主DAG文件 ├── plugins/ # 自定义Operator或Hook ├── scripts/ # 独立的SQL或Shell脚本 ├── sql/ │ └── transform_user_behavior.sql └── config/ └── constants.py # 存放配置常量DAG代码实现(user_behavior_etl.py)from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator, BranchPythonOperator from airflow.operators.bash import BashOperator from airflow.operators.email import EmailOperator from airflow.operators.dummy import DummyOperator from airflow.providers.http.sensors.http import HttpSensor from airflow.providers.postgres.operators.postgres import PostgresOperator from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator # 假设我们使用社区提供的ClickHouse Provider from airflow.providers.clickhouse.operators.clickhouse import ClickHouseOperator from airflow.utils.edgemodifier import Label import sys sys.path.append(/path/to/my_data_pipelines) from config.constants import API_ENDPOINT, EMAIL_LIST default_args { owner: bi_team, depends_on_past: False, email_on_failure: True, email_on_retry: False, retries: 3, retry_delay: timedelta(minutes2), execution_timeout: timedelta(hours2), } def _extract_api_data(**context): 从API提取数据并推送到XCom供下游任务使用 import requests import pandas as pd execution_date context[execution_date] # 构造请求参数例如获取前一天的数据 params {date: (execution_date - timedelta(days1)).strftime(%Y-%m-%d)} response requests.get(API_ENDPOINT, paramsparams) response.raise_for_status() data response.json() # 简单转换这里可以更复杂 df pd.DataFrame(data[records]) # 将数据以JSON字符串形式推送到XCom context[ti].xcom_push(keyapi_raw_data, valuedf.to_json(orientrecords)) return fExtracted {len(df)} records. def _process_data(**context): 处理数据清洗、去重、转换 import pandas as pd import json ti context[ti] raw_data_json ti.xcom_pull(task_idsextract_from_api, keyapi_raw_data) df pd.read_json(raw_data_json, orientrecords) # 执行清洗逻辑例如去重、处理空值、类型转换 df_cleaned df.drop_duplicates(subset[user_id, event_time]) df_cleaned[event_time] pd.to_datetime(df_cleaned[event_time]) # 再次推送到XCom或者也可以直接写入临时存储如S3/MinIO ti.xcom_push(keycleaned_data, valuedf_cleaned.to_json(orientrecords)) return fProcessed {len(df_cleaned)} records. def _check_data_quality(**context): 数据质量检查决定是否继续执行加载任务 import pandas as pd import json ti context[ti] cleaned_data_json ti.xcom_pull(task_idsprocess_and_clean, keycleaned_data) df pd.read_json(cleaned_data_json, orientrecords) # 简单的质量检查规则 if df.empty: return alert_empty_data # 返回下游任务的task_id elif df[user_id].isnull().sum() len(df) * 0.1: # 超过10%的user_id为空 return alert_bad_quality else: return load_to_warehouse with DAG( dag_iduser_behavior_daily_etl, default_argsdefault_args, description每日用户行为日志ETL流水线, schedule_interval30 1 * * *, # 每天凌晨1:30运行 start_datedatetime(2024, 1, 1), catchupFalse, max_active_runs1, # 防止并发导致的数据混乱 tags[production, etl, user_behavior], ) as dag: start DummyOperator(task_idstart) end DummyOperator(task_idend) # 1. 检查API是否可用传感器任务 api_available HttpSensor( task_idapi_available, http_conn_iduser_behavior_api, # 需要在Airflow UI中预先配置Connection endpoint/health, response_checklambda response: response.status_code 200, poke_interval30, timeout5*60, modepoke, ) # 2. 提取数据 extract_from_api PythonOperator( task_idextract_from_api, python_callable_extract_api_data, ) # 3. 清洗处理 process_and_clean PythonOperator( task_idprocess_and_clean, python_callable_process_data, ) # 4. 数据质量检查与分支 data_quality_check BranchPythonOperator( task_iddata_quality_check, python_callable_check_data_quality, ) # 5. 分支路径正常加载 load_to_warehouse ClickHouseOperator( task_idload_to_warehouse, clickhouse_conn_idclickhouse_prod, sql INSERT INTO dw.user_behavior_daily SELECT user_id, event_type, event_time, device_id, ... FROM input(user_id String, event_type String, ...) FORMAT JSONEachRow , parameters{data: {{ ti.xcom_pull(task_idsprocess_and_clean, keycleaned_data) }}}, ) # 6. 分支路径告警 alert_empty_data EmailOperator( task_idalert_empty_data, toEMAIL_LIST, subjectAirflow Alert: 用户行为数据为空 - {{ ds }}, html_contentp今日用户行为API返回数据为空请检查数据源。/p, ) alert_bad_quality EmailOperator( task_idalert_bad_quality, toEMAIL_LIST, subjectAirflow Alert: 用户行为数据质量异常 - {{ ds }}, html_contentp今日用户行为数据质量检查未通过如大量空值请核查。/p, ) # 7. 后续步骤数据聚合仅当成功加载后执行 run_daily_aggregation PostgresOperator( task_idrun_daily_aggregation, postgres_conn_idpostgres_warehouse, sqlsql/transform_user_behavior.sql, # 引用外部SQL文件 ) send_daily_report EmailOperator( task_idsend_daily_report, toEMAIL_LIST, subject每日用户行为报告 - {{ ds }}, html_contentpETL任务已完成详细报告见附件或BI系统。/p, ) # 定义依赖关系 start api_available extract_from_api process_and_clean data_quality_check # 分支依赖 data_quality_check [load_to_warehouse, alert_empty_data, alert_bad_quality] # 主路径继续 load_to_warehouse run_daily_aggregation send_daily_report end # 告警路径直接结束 alert_empty_data end alert_bad_quality end4.2 代码详解与避坑指南使用XCom进行小数据量传递XCom是Airflow中任务间传递数据的机制。但请注意XCom不适合传递大型数据集如几百MB的数据因为数据会存储在元数据库中。对于大数据最佳实践是将中间数据写入共享存储如S3、HDFS、NFS然后传递文件路径。上述示例仅用于演示小数据场景。连接管理代码中的http_conn_id、postgres_conn_id等需要在Airflow Web UI的Admin - Connections中预先配置。这是集中管理凭证和连接信息的安全方式避免将密码硬编码在DAG中。模板变量Airflow支持Jinja2模板。{{ ds }}是内置变量代表当前执行日期的“YYYY-MM-DD”格式字符串。在EmailOperator的标题中使用它可以让告警邮件自带日期上下文非常实用。BranchPythonOperator这是一个强大的操作符允许你根据上游任务的输出动态决定下游执行路径。它返回的是下一个要执行的task_id。这实现了工作流中的条件逻辑。执行超时与重试在default_args中设置了execution_timeout和retries。对于长时间运行的任务如Spark作业需要合理设置超时时间。重试机制配合retry_delay可以应对短暂的网络抖动或资源不足。依赖关系清晰化使用和运算符让依赖关系一目了然。对于复杂的依赖可以使用set_upstream和set_downstream方法但位运算符更简洁。一个真实的坑时区问题。Airflow默认使用UTC时间。如果你的业务系统使用本地时间如东八区在定义schedule_interval和start_date时以及在任务逻辑中处理execution_date时必须非常小心。建议在DAG级别或全局配置中统一时区并在所有时间处理逻辑中显式进行时区转换。5. 进阶实战性能调优、监控与运维最佳实践当你的Airflow集群开始承载成百上千个DAG和任务时性能和运维就成了挑战。以下是一些来自实战的经验。5.1 性能调优关键点调度器性能[scheduler] parsing_processes增加DAG解析进程数加快DAG文件解析速度。通常设置为CPU核心数。[scheduler] max_threads增加调度器主线程数用于处理任务调度逻辑。减少DAG文件的复杂度避免在DAG文件的顶层模块中执行耗时的操作如大数据量查询、网络请求。将这些逻辑移到Operator内部或辅助函数中。调度器会频繁导入DAG文件复杂的顶层代码会成为瓶颈。使用.airflowignore文件忽略不需要被扫描的目录或文件减少调度器负担。Executor与WorkerCelery Worker并发度通过-c参数或配置worker_concurrency调整每个Worker可以同时执行的任务数。不要超过Worker机器的CPU核心数。任务资源隔离对于资源消耗大CPU/内存的任务使用KubernetesPodOperator或DockerOperator进行隔离避免影响其他任务。也可以使用Celery的任务队列将重型任务路由到专用的高配Worker节点。优化任务粒度避免设计运行时间过长的“巨无霸”任务。将其拆分为多个小的、可重试的原子任务。这提高了并行度和失败恢复的灵活性。数据库优化Airflow的元数据库在运行中会产生大量记录TaskInstance, Log, XCom等。必须定期清理。使用Airflow内置的airflow db clean命令2.0版本后或维护DAG来删除旧数据。否则数据库会无限膨胀导致性能急剧下降。对于PostgreSQL考虑对task_instance、log等大表建立合适的索引如dag_id,state,execution_date。5.2 监控与告警体系“没有监控的系统就是在裸奔。” Airflow的监控分为几个层次内置UI监控Tree View、Graph View、Gantt Chart是基本的任务状态和运行时长可视化工具。重点关注长时间运行、失败或重试状态的任务。日志管理Airflow默认将任务日志存储在本地文件系统。生产环境应集成到集中式日志系统如ELK Stack、LokiGraylog。配置[logging] remote_logging相关设置将日志发送到远程存储如S3、GCS并供日志系统抓取。这样可以在一个地方查看所有任务的日志方便排错。指标暴露Airflow内置了StatsD指标可以集成到PrometheusGrafana中。关键指标包括scheduler_heartbeat调度器是否存活。executor.open_slots/executor.queued_tasksExecutor资源使用情况。dagrun.duration.*/dagrun.schedule_delay.*DAG运行时长和调度延迟。task_instance.*任务状态计数成功、失败、重试。 在Grafana中绘制这些指标的仪表盘可以实时掌握集群健康度。主动告警任务失败告警利用email_on_failure参数是最基本的。更高级的做法是使用on_failure_callback在任务失败时触发一个自定义的Python函数可以集成到企业微信、钉钉、Slack、PagerDuty等告警平台。DAG运行超时告警使用on_retry_callback或专门的监控DAG来检查其他DAG的运行状态。可以编写一个传感器DAG定期检查关键DAG的最新运行状态是否成功、是否按时完成。Scheduler/Worker进程存活告警通过监控系统如Prometheus的up指标或简单的HTTP探针监控Airflow各个服务的进程。5.3 版本控制与CI/CD将DAG代码视为应用程序代码纳入Git版本控制。并建立CI/CD流水线代码检查在CI阶段运行pylint、black、mypy等工具确保代码风格和质量。单元测试使用Airflow的测试工具如airflow.utils.dag_cycle_tester检查循环依赖和pytest对自定义的Python函数、Operator进行单元测试。集成测试可以搭建一个与生产环境隔离的测试Airflow环境将DAG部署上去触发一次运行验证整个流程是否通畅。自动部署CD阶段将通过测试的DAG代码同步到生产环境的DAGS_FOLDER目录。可以使用rsync、K8s ConfigMap或专门的部署工具。一个至关重要的经验DAG的幂等性与数据日期。在设计DAG时必须确保它能够处理“重跑”场景。这意味着任务逻辑应该基于execution_date或data_interval_start/end来确定它要处理的数据范围而不是“今天”或“现在”。这样当你需要重新处理某一天的历史数据时只需要清除那一天的DAG Run状态并重新触发即可任务会自动处理正确的数据切片而不会影响到其他日期的数据。这是构建健壮数据流水线的基石。Airflow不是一个“安装即忘”的工具它是一个需要精心设计和持续运维的数据编排平台。从简单的任务调度起步逐步深入到依赖管理、错误处理、性能优化和监控告警你会逐渐构建起一个可靠、高效、可维护的数据基础设施。这个过程充满挑战但当你看到复杂的数据流在Airflow的调度下井然有序地自动运行时那种掌控感和效率提升会让所有的投入都变得值得。
