1. 项目概述今天要分享的是如何用Python在Airflow中快速定义定时数据清洗任务。作为一名长期从事数据工程的老兵我深知数据清洗在ETL流程中的重要性。Airflow作为业界广泛使用的工作流调度工具配合Python的灵活性能够帮我们高效解决定时数据清洗的痛点。这个教程将带你3分钟快速上手实现以下核心功能用Python定义DAG有向无环图设置定时调度策略实现数据清洗任务的自动化执行监控任务执行状态2. 核心概念解析2.1 Airflow基础架构Airflow的核心组件包括Web Server可视化任务监控界面Scheduler任务调度核心Executor任务执行器Metadata Database存储任务元数据DAG Directory存放DAG定义文件2.2 DAG定义关键参数from datetime import datetime from airflow import DAG default_args { owner: data_team, start_date: datetime(2023, 1, 1), retries: 3, } dag DAG( data_cleaning, default_argsdefault_args, schedule_interval0 3 * * *, # 每天凌晨3点执行 catchupFalse )3. 环境准备3.1 安装Airflowpip install apache-airflow airflow db init airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com3.2 目录结构~/airflow/ ├── dags/ # 存放DAG文件 ├── logs/ # 任务执行日志 ├── airflow.cfg # 配置文件 └── airflow.db # SQLite数据库4. 数据清洗任务实现4.1 定义PythonOperatorfrom airflow.operators.python import PythonOperator def clean_data(**context): import pandas as pd # 从上下文获取执行日期 exec_date context[execution_date] # 数据清洗逻辑 raw_data pd.read_csv(f/data/raw/{exec_date}.csv) cleaned_data raw_data.dropna().drop_duplicates() cleaned_data.to_csv(f/data/cleaned/{exec_date}.csv, indexFalse) clean_task PythonOperator( task_idclean_data, python_callableclean_data, provide_contextTrue, dagdag )4.2 完整DAG示例from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator default_args { owner: data_engineer, depends_on_past: False, email_on_failure: True, retries: 2, retry_delay: timedelta(minutes5) } def clean_function(**context): # 实际清洗逻辑 pass with DAG( daily_data_cleaning, default_argsdefault_args, schedule_interval0 3 * * *, start_datedatetime(2023, 1, 1), catchupFalse ) as dag: clean_task PythonOperator( task_idclean_and_transform, python_callableclean_function, provide_contextTrue ) # 可以添加更多任务 # notify EmailOperator(...) # clean_task notify5. 高级配置技巧5.1 参数化DAGdef create_dag(dag_id, schedule, default_args): with DAG(dag_id, schedule_intervalschedule, default_argsdefault_args) as dag: t1 PythonOperator( task_idclean_ dag_id, python_callableclean_data ) return dag # 批量创建多个DAG for n in range(1, 5): dag_id fcleaning_workflow_{n} globals()[dag_id] create_dag( dag_iddag_id, scheduledaily, default_argsdefault_args )5.2 使用XCom跨任务通信def extract(**context): # 提取数据 data [...] context[ti].xcom_push(keyraw_data, valuedata) def transform(**context): # 获取上游数据 raw_data context[ti].xcom_pull(task_idsextract, keyraw_data) # 转换逻辑 cleaned_data [x for x in raw_data if x is not None] return cleaned_data6. 生产环境最佳实践6.1 错误处理机制from airflow.exceptions import AirflowFailException def clean_data(**context): try: # 清洗逻辑 if error_condition: raise AirflowFailException(数据质量检查失败) except Exception as e: log_error(e) raise6.2 资源控制clean_task PythonOperator( task_idclean_large_dataset, python_callableclean_data, executor_config{ KubernetesExecutor: { request_memory: 4Gi, limit_memory: 8Gi } } )7. 监控与告警7.1 SLA配置dag DAG( critical_cleaning, sla_miss_callbacknotify_sla_miss, default_args{ sla: timedelta(hours1) } )7.2 自定义指标监控from airflow.models import Variable def track_metrics(**context): processed_rows 1000 Variable.set(last_run_metrics, {rows: processed_rows, ts: context[ts]})8. 性能优化8.1 任务并行化from airflow.utils.task_group import TaskGroup with TaskGroup(parallel_cleaning) as cleaning_group: clean_task1 PythonOperator(task_idclean_source1, ...) clean_task2 PythonOperator(task_idclean_source2, ...) # 这两个任务将并行执行 [clean_task1, clean_task2] join_task8.2 增量处理模式def incremental_clean(**context): last_run Variable.get(last_success_run) new_data get_data_since(last_run) # 处理增量数据9. 常见问题排查9.1 任务未按预期调度检查要点start_date是否设置正确schedule_interval语法是否正确DAG文件是否有语法错误Scheduler是否正常运行9.2 任务执行失败调试步骤查看任务日志检查依赖项版本验证输入数据测试独立Python脚本10. 扩展应用场景10.1 数据质量检查from airflow.providers.postgres.operators.postgres import PostgresOperator validate_task PostgresOperator( task_idvalidate_data, sql SELECT COUNT(*) FROM cleaned_data WHERE last_update {{ execution_date }} , postgres_conn_idpostgres_conn )10.2 机器学习管道train_task PythonOperator( task_idtrain_model, python_callabletrain_with_cleaned_data, op_kwargs{ data_path: /data/cleaned/{{ ds }}.csv } )在实际项目中我发现将清洗逻辑模块化非常重要。我通常会创建一个独立的Python包来存放所有数据转换逻辑然后在Airflow的PythonOperator中调用这些函数。这样的架构既保持了DAG文件的简洁又便于单独测试数据清洗逻辑。另一个实用技巧是在开发阶段设置catchupFalse避免意外触发大量历史任务。等DAG稳定后再根据需要开启补跑功能。
1. 项目概述今天要分享的是如何用Python在Airflow中快速定义定时数据清洗任务。作为一名长期从事数据工程的老兵我深知数据清洗在ETL流程中的重要性。Airflow作为业界广泛使用的工作流调度工具配合Python的灵活性能够帮我们高效解决定时数据清洗的痛点。这个教程将带你3分钟快速上手实现以下核心功能用Python定义DAG有向无环图设置定时调度策略实现数据清洗任务的自动化执行监控任务执行状态2. 核心概念解析2.1 Airflow基础架构Airflow的核心组件包括Web Server可视化任务监控界面Scheduler任务调度核心Executor任务执行器Metadata Database存储任务元数据DAG Directory存放DAG定义文件2.2 DAG定义关键参数from datetime import datetime from airflow import DAG default_args { owner: data_team, start_date: datetime(2023, 1, 1), retries: 3, } dag DAG( data_cleaning, default_argsdefault_args, schedule_interval0 3 * * *, # 每天凌晨3点执行 catchupFalse )3. 环境准备3.1 安装Airflowpip install apache-airflow airflow db init airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email adminexample.com3.2 目录结构~/airflow/ ├── dags/ # 存放DAG文件 ├── logs/ # 任务执行日志 ├── airflow.cfg # 配置文件 └── airflow.db # SQLite数据库4. 数据清洗任务实现4.1 定义PythonOperatorfrom airflow.operators.python import PythonOperator def clean_data(**context): import pandas as pd # 从上下文获取执行日期 exec_date context[execution_date] # 数据清洗逻辑 raw_data pd.read_csv(f/data/raw/{exec_date}.csv) cleaned_data raw_data.dropna().drop_duplicates() cleaned_data.to_csv(f/data/cleaned/{exec_date}.csv, indexFalse) clean_task PythonOperator( task_idclean_data, python_callableclean_data, provide_contextTrue, dagdag )4.2 完整DAG示例from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator default_args { owner: data_engineer, depends_on_past: False, email_on_failure: True, retries: 2, retry_delay: timedelta(minutes5) } def clean_function(**context): # 实际清洗逻辑 pass with DAG( daily_data_cleaning, default_argsdefault_args, schedule_interval0 3 * * *, start_datedatetime(2023, 1, 1), catchupFalse ) as dag: clean_task PythonOperator( task_idclean_and_transform, python_callableclean_function, provide_contextTrue ) # 可以添加更多任务 # notify EmailOperator(...) # clean_task notify5. 高级配置技巧5.1 参数化DAGdef create_dag(dag_id, schedule, default_args): with DAG(dag_id, schedule_intervalschedule, default_argsdefault_args) as dag: t1 PythonOperator( task_idclean_ dag_id, python_callableclean_data ) return dag # 批量创建多个DAG for n in range(1, 5): dag_id fcleaning_workflow_{n} globals()[dag_id] create_dag( dag_iddag_id, scheduledaily, default_argsdefault_args )5.2 使用XCom跨任务通信def extract(**context): # 提取数据 data [...] context[ti].xcom_push(keyraw_data, valuedata) def transform(**context): # 获取上游数据 raw_data context[ti].xcom_pull(task_idsextract, keyraw_data) # 转换逻辑 cleaned_data [x for x in raw_data if x is not None] return cleaned_data6. 生产环境最佳实践6.1 错误处理机制from airflow.exceptions import AirflowFailException def clean_data(**context): try: # 清洗逻辑 if error_condition: raise AirflowFailException(数据质量检查失败) except Exception as e: log_error(e) raise6.2 资源控制clean_task PythonOperator( task_idclean_large_dataset, python_callableclean_data, executor_config{ KubernetesExecutor: { request_memory: 4Gi, limit_memory: 8Gi } } )7. 监控与告警7.1 SLA配置dag DAG( critical_cleaning, sla_miss_callbacknotify_sla_miss, default_args{ sla: timedelta(hours1) } )7.2 自定义指标监控from airflow.models import Variable def track_metrics(**context): processed_rows 1000 Variable.set(last_run_metrics, {rows: processed_rows, ts: context[ts]})8. 性能优化8.1 任务并行化from airflow.utils.task_group import TaskGroup with TaskGroup(parallel_cleaning) as cleaning_group: clean_task1 PythonOperator(task_idclean_source1, ...) clean_task2 PythonOperator(task_idclean_source2, ...) # 这两个任务将并行执行 [clean_task1, clean_task2] join_task8.2 增量处理模式def incremental_clean(**context): last_run Variable.get(last_success_run) new_data get_data_since(last_run) # 处理增量数据9. 常见问题排查9.1 任务未按预期调度检查要点start_date是否设置正确schedule_interval语法是否正确DAG文件是否有语法错误Scheduler是否正常运行9.2 任务执行失败调试步骤查看任务日志检查依赖项版本验证输入数据测试独立Python脚本10. 扩展应用场景10.1 数据质量检查from airflow.providers.postgres.operators.postgres import PostgresOperator validate_task PostgresOperator( task_idvalidate_data, sql SELECT COUNT(*) FROM cleaned_data WHERE last_update {{ execution_date }} , postgres_conn_idpostgres_conn )10.2 机器学习管道train_task PythonOperator( task_idtrain_model, python_callabletrain_with_cleaned_data, op_kwargs{ data_path: /data/cleaned/{{ ds }}.csv } )在实际项目中我发现将清洗逻辑模块化非常重要。我通常会创建一个独立的Python包来存放所有数据转换逻辑然后在Airflow的PythonOperator中调用这些函数。这样的架构既保持了DAG文件的简洁又便于单独测试数据清洗逻辑。另一个实用技巧是在开发阶段设置catchupFalse避免意外触发大量历史任务。等DAG稳定后再根据需要开启补跑功能。
双梁正问题 【 torch 实战】（Python代码实现）)
