# Apache Airflow 编排掌握 Apache Airflow 工作流程编排的综合技能。该技能涵盖 DAG 开发、操作员、传感器、任务依赖性、动态工作流程、XCom 通信、调度模式和生产部署策略。 ## 何时使用此技能在以下情况下使用此技能： * 构建和管理具有任务依赖性的复杂数据管道 * 跨多个系统编排 ETL/ELT 工作流程 * 调度和监控批处理作业 * 协调多步骤数据转换 * 通过条件执行和分支管理工作流程 * 实施事件驱动或基于资产的工作流程 * 部署生产级工作流程自动化 * 创建以编程方式生成任务的动态工作流程 * 协调跨集群的分布式任务执行 * 通过工作流编排构建数据工程平台 ## 核心概念 ### 什么是 Apache Airflow？ Apache Airflow 是一个开源平台，用于以编程方式创作、调度和监控工作流程。它允许您使用 Python 代码将工作流程定义为有向无环图 (DAG)，从而使复杂的工作流程编排可维护且受版本控制。 **关键原则：** * **动态** ：工作流程在Python中定义，支持动态生成 * **可扩展**：丰富的操作员、传感器和钩子生态系统 * **可扩展** ：可以从单机扩展到大型集群 * **Observable** ：用于监控和故障排除的综合 UI ### DAG（有向无环图） DAG 是为了反映其关系和依赖关系而组织的任务的集合。 **DAG 属性：** * **dag_id** ：DAG 的唯一标识符 * **start_date** ：DAG 应开始安排的时间 * **计划** ：运行频率（cron、timedelta 或基于资产） * **catchup** ：是否在 DAG 激活时运行错过的间隔 * **tags** ：用于组织和过滤的标签 * **default_args** : DAG 中所有任务的默认参数 **DAG 定义示例：** 从日期时间导入日期时间从airflow.sdk导入DAG 与 DAG( dag_id="example_dag", 开始日期 = 日期时间(2022, 1, 1), Schedule="0 0 * * *", # 每天午夜追赶=假，标签=["示例","教程"], ) 作为达格： # 此处定义的任务通过 ### 任务和操作员 **任务**是 Airflow 中执行的基本单元。 **运算符**是用于创建任务的模板。 **常见运算符类型：** 1. **BashOperator** ：执行bash命令 2. **PythonOperator** ：执行Python函数 3. **EmailOperator** : 发送电子邮件 4. **EmptyOperator** ：占位符/虚拟任务 5. **自定义运算符**：针对特定需求的用户定义运算符 **操作员与任务：** * 运算符：模板/类定义 * 任务：实例化具有特定参数的算子 ### 任务依赖关系任务依赖关系定义了执行顺序和工作流结构。 **依赖运算符：** * `>>`：设置下游依赖关系（任务1 >> 任务2） * `<<`: 设置上游依赖关系 (task2 << task1) * `chain()`：多个任务的顺序依赖 * `cross_downstream()`：多对多关系 **依赖关系示例：** # 简单的线性流任务1 >> 任务2 >> 任务3 # 扇出模式任务1 >> [任务2，任务3，任务4] # 扇入模式 [任务1，任务2，任务3] >> 任务4 # 复杂的依赖关系第一个任务 >> [第二个任务，第三个任务] 第三任务 << 第四任务 ### 执行者执行者决定任务的运行方式和位置。 **执行者类型：** * **SequentialExecutor** ：单线程，本地（默认，不适用于生产） * **LocalExecutor** ：多线程，单机 * **CeleryExecutor** ：使用 Celery 进行分布式执行 * **KubernetesExecutor** ：每个任务在单独的 Kubernetes pod 中运行 * **DaskExecutor** ：使用Dask分布式执行 ### 调度程序气流调度器： * 监控所有 DAG 及其任务 * 根据依赖关系和时间表触发任务实例 * 提交任务给执行者执行 * 处理重试和任务状态管理 **启动调度程序：** 气流调度器 ## DAG 开发模式 ### 基本 DAG 结构每个 DAG 都遵循以下结构：从日期时间导入日期时间从airflow.sdk导入DAG 从airflow.providers.standard.operators.bash导入BashOperator 与 DAG( dag_id="basic_dag", 开始日期 = 日期时间(2022, 1, 1), 时间表=“0 0 * * *”，追赶=假， ) 作为达格：任务1 = BashOperator( 任务id=“任务1”， bash_command="echo '任务 1 已执行'" ）任务2 = BashOperator( 任务id=“任务2”， bash_command="echo '任务2已执行'" ）任务1 >> 任务2 ### 任务依赖关系和链 **直线链：** 来自airflow.sdk导入链 # 这些是等价的：任务1 >> 任务2 >> 任务3 >> 任务4 链（任务1，任务2，任务3，任务4） **动态链：** 来自airflow.sdk导入链从airflow.operators.empty导入EmptyOperator # 动态生成并链接任务链(*[EmptyOperator(task_id=f"task_{i}") for i in range(1, 6)]) **成对链：** 来自airflow.sdk导入链 # 创建配对依赖项： # op1 >> op2 >> op4 >> op6 # op1 >> op3 >> op5 >> op6 链（op1，[op2，op3]，[op4，op5]，op6） **跨下游：** 从airflow.sdk导入cross_downstream # op1 和 op2 都输入到 op3 和 op4 中 cross_downstream([op1, op2], [op3, op4]) ### 分支和条件执行 **分支Python操作符：** 从airflow.operators.python导入BranchPythonOperator def select_branch(**context): 如果上下文['data_interval_start'].day == 1：返回“每月任务” 返回“每日任务” 分支 = BranchPythonOperator( task_id='branch_task', python_callable=选择分支） daily_task = BashOperator(task_id='daily_task', bash_command='echo daily') Monthly_task = BashOperator(task_id='monthly_task', bash_command='回显每月') 分支 >> [每日任务、每月任务] **自定义分支运营商：** 从airflow.operators.branch导入BaseBranchOperator 类 MyBranchOperator(BaseBranchOperator): def select_branch(self, 上下文): ”“” 在每月的第一天运行额外的分支 ”“” 如果上下文['data_interval_start'].day == 1：返回['每日任务ID'，'每月任务ID'] elif context['data_interval_start'].day == 2: 返回“每日任务 ID” 其他： return None # 跳过所有下游任务 ### 组织任务组任务组有助于分层组织相关任务：从airflow.sdk导入task_group 从airflow.operators.empty导入EmptyOperator @任务组() def data_processing_group(): 提取 = EmptyOperator(task_id="提取") 变换 = EmptyOperator(task_id="transform") 负载 = EmptyOperator(task_id="负载") 提取>>转换>>加载 @任务组() defvalidation_group(): validate_schema = EmptyOperator(task_id="validate_schema") validate_data = EmptyOperator(task_id="validate_data") 验证模式 >> 验证数据开始 = EmptyOperator(task_id="开始") 结束 = EmptyOperator(task_id="结束") 开始 >> data_processing_group() >> valid_group() >> 结束 ### 边缘标签为了清晰起见，向依赖边添加标签：从airflow.sdk导入标签 # 内联标签 my_task >> 标签("空时") >> other_task # 基于方法的标记 my_task.set_downstream(other_task, Label("空时")) ### LastOnlyOperator 如果不是最新的 DAG 运行则跳过任务：从airflow.operators.latest_only导入LatestOnlyOperator 从airflow.operators.empty导入EmptyOperator 进口摆锤与 DAG( dag_id='latest_only_example', start_date=pendulum.datetime(2023, 1, 1, tz="UTC"), 追赶=真，时间表=“@每日”， ) 作为达格： latest_only = LatestOnlyOperator(task_id='latest_only') 任务1 = EmptyOperator(task_id='任务1') 任务2 = EmptyOperator(task_id='任务2') 任务3 = EmptyOperator(task_id='task3') 任务4 = EmptyOperator(task_id='task4',trigger_rule='all_done') 最新_仅 >> 任务1 >> 任务3 最新_仅 >> 任务4 任务2 >> 任务3 任务2 >> 任务4 ## 运营商深入探究 ### Bash 操作符执行bash命令：从airflow.providers.standard.operators.bash导入BashOperator bash_task = BashOperator( task_id =“bash_example”， bash_command="echo '来自 Bash 的问候';日期", env={'MY_VAR': 'value'}, # 环境变量 append_env=True, # 附加到现有环境变量输出编码='utf-8' ） **复杂的 Bash 命令：** bash_complex = BashOperator( task_id =“complex_bash”， bash_命令=“”“ cd /路径/到/目录 python process_data.py --输入{{ds}} --输出{{today_ds}} 如果 [ $? -eq 0]；然后回显“成功” 否则 echo "失败" && 退出 1 菲 """, ） ### Python 运算符执行Python函数：从airflow.providers.standard.operators.python导入PythonOperator def my_python_function(名称, **上下文): print(f"你好{name}！") print(f"执行日期：{context['ds']}") 返回“成功” python_task = PythonOperator( task_id =“python_example”， python_callable=my_python_function, op_kwargs={'name': '气流'}, 提供_上下文=真） **使用 PythonOperator 的传统 ETL：** 导入 json 进口摆锤从airflow.sdk导入DAG 从airflow.providers.standard.operators.python导入PythonOperator def 提取（）： data_string = '{“1001”：301.27，“1002”：433.21，“1003”：502.22}' 返回 json.loads(data_string) 定义变换（ti）： # 从 XCom 拉取 order_data_dict = ti.xcom_pull(task_ids="extract") 总订单值 = sum(order_data_dict.values()) 返回{“total_order_value”：total_order_value} 定义负载（ti）： # 从 XCom 拉取总计 = ti.xcom_pull(task_ids="transform")["total_order_value"] print(f"订单总金额为：{total:.2f}") 与 DAG( dag_id="legacy_etl_pipeline", 时间表=无， start_date=pendulum.datetime(2021, 1, 1, tz="UTC"), 追赶=假， ) 作为达格： extract_task = PythonOperator(task_id="extract", python_callable=extract) Transform_task = PythonOperator(task_id="transform", python_callable=transform) load_task = PythonOperator(task_id="load", python_callable=load) 提取任务>>转换任务>>加载任务 ### 电子邮件操作员发送电子邮件通知：从airflow.providers.smtp.operators.smtp导入EmailOperator email_task = EmailOperator( task_id='发送电子邮件', to='recipient@example.com', subject='气流通知', html_content='

任务成功完成！

', cc=['cc@example.com'], bcc=['bcc@example.com'] ） ### 空运算符工作流结构的占位符：从airflow.operators.empty导入EmptyOperator 开始 = EmptyOperator(task_id='开始') 结束 = EmptyOperator(task_id='结束') # 对于组织复杂的 DAG 很有用开始 >> [任务1, 任务2, 任务3] >> 结束 ### 自定义运算符创建可重用的自定义运算符：从airflow.models导入BaseOperator 从airflow.utils.decorators导入apply_defaults 类 MyCustomOperator(BaseOperator): @应用默认值 def __init__(self, my_param, *args, **kwargs): super().__init__(*args, **kwargs) self.my_param = my_param def 执行（自身，上下文）： self.log.info(f"使用参数执行：{self.my_param}") # 此处自定义逻辑返回“结果” # 用法自定义任务 = MyCustomOperator( task_id="自定义", 我的参数=“值” ） ## 传感器深入探讨传感器是一种特殊类型的操作器，它会等待满足特定条件后再继续操作。 ### 外部任务传感器等待其他 DAG 中的任务：从airflow.providers.standard.sensors.external_task导入ExternalTaskSensor 进口摆锤与 DAG( dag_id="example_external_task_sensor", start_date=pendulum.datetime(2021, 10, 20, tz="UTC"), 追赶=假，时间表=无， ) 作为达格： wait_for_task = 外部任务传感器( task_id =“wait_for_task”， external_dag_id =“upstream_dag”， external_task_id="upstream_task", allowed_states=["成功"], failed_states=["失败"], execution_delta=None, # 相同的execution_date 超时=600, # 10 分钟 poke_interval=60, # 每 60 秒检查一次） **可延迟的外部任务传感器：** # 更高效 - 在等待时释放工作槽 wait_for_task_async = 外部任务传感器（ task_id =“wait_for_task_async”， external_dag_id =“upstream_dag”， external_task_id="upstream_task", allowed_states=["成功"], failed_states=["失败"], deferrable=True, # 使用异步模式） ### 文件传感器等待文件出现：从airflow.sensors.filesystem导入FileSensor 等待文件=文件传感器（ task_id="wait_for_file", 文件路径=“/path/to/file.csv”，戳_间隔=30，超时=600， mode='poke' # 或'重新安排'长时间等待） ### 时间增量传感器等待特定时间段：从日期时间导入时间增量从airflow.sensors.time_delta导入TimeDeltaSensor wait_one_hour = TimeDeltaSensor( task_id =“等待一小时”，增量=时间增量(小时=1) ） ### BigQuery 表传感器等待 BigQuery 表存在：从 airflow.providers.google.cloud.sensors.bigquery 导入 BigQueryTableExistenceSensor 进口摆锤与 DAG( dag_id="bigquery_sensor_example", start_date=pendulum.datetime(2023, 10, 26, tz="UTC"), ) 作为达格： wait_for_table = BigQueryTableExistenceSensor( task_id="wait_for_table", project_id="您的项目 ID", dataset_id="你的数据集", table_id="你的表", bigquery_conn_id="google_cloud_default", 位置=“美国”， poke_间隔=60，超时=3600，） ### 定制传感器针对特定条件创建自定义传感器：从airflow.sensors.base导入BaseSensorOperator 从airflow.utils.decorators导入apply_defaults 类 MyCustomSensor(BaseSensorOperator): @应用默认值 def __init__(self, my_condition, *args, **kwargs): super().__init__(*args, **kwargs) self.my_condition = 我的_condition def poke(自身，上下文)： # 满足条件时返回True self.log.info(f"检查条件：{self.my_condition}") # 自定义逻辑来检查条件返回 check_condition(self.my_condition) ### 可延迟传感器可延迟传感器在等待时释放工作槽：从日期时间导入时间增量从airflow.sdk导入BaseSensorOperator，StartTriggerArgs 类 WaitHoursSensor(BaseSensorOperator): start_trigger_args = StartTriggerArgs( trigger_cls =“airflow.providers.standard.triggers.temporal.TimeDeltaTrigger”， trigger_kwargs={"时刻": timedelta(小时=1)}, next_method =“执行完成”， next_kwargs=无，超时=无，）从触发开始 = True def __init__(self，*args，trigger_kwargs=无，start_from_trigger=True，**kwargs)： super().__init__(*args, **kwargs) 如果trigger_kwargs： self.start_trigger_args.trigger_kwargs = trigger_kwargs self.start_from_trigger = start_from_trigger defexecute_complete（自身，上下文，事件=无）： return # 任务完成 ## XComs（交叉通信） XCom 通过存储和检索数据来实现任务间通信。 ### XCom 基本用法 **推送至 XCom：** def push_function(**上下文): 值=“重要数据” context['ti'].xcom_push(key='my_key', value=value) # 或者简单地返回（使用“return_value”键）返回值 Push_task = PythonOperator( task_id='推送', python_callable=push_function, 提供_上下文=真） **从 XCom 拉取：** def pull_function(**上下文): # 按task_id拉取（使用'return_value'键）值 = context['ti'].xcom_pull(task_ids='push') # 使用特定的 key 拉取 value = context['ti'].xcom_pull(task_ids='push', key='my_key') print(f"拉取的值：{value}") pull_task = PythonOperator( task_id='拉', python_callable=pull_function, 提供_上下文=真） ### XCom 与 TaskFlow API TaskFlow API 自动管理 XCom：从airflow.decorators导入任务 @任务 def 提取（）：返回{“数据”：[1,2,3,4,5]} @任务 def 变换（data_dict）： # 自动从extract接收XCom 总计 = sum(data_dict['data']) 返回{“总计”：总计} @任务定义负载（摘要）： print(f"总计：{summary['total']}") # 自动 XCom 处理数据=提取（）摘要 = 变换（数据）负载（摘要） ### XCom 最佳实践 **尺寸限制：** * XComs存储在元数据数据库中 * 保持 XCom 数据较小（建议 < 1MB） * 对于大数据，存储在外部系统中并传递引用 **外部存储示例：** @任务 def process_large_data(): # 处理数据大结果=计算大数据集（） # 存储在S3/GCS中 file_path = save_to_s3(large_result, "s3://bucket/result.parquet") # 只返回路径返回{“结果路径”：文件路径} @任务 def Consumer_large_data（元数据）： # 从 S3/GCS 加载数据 = load_from_s3(元数据['结果路径']) 过程（数据） ### XCom 与操作员 **阅读模板中的 XCom：** 从airflow.providers.standard.operators.bash导入BashOperator 进程文件 = BashOperator( task_id=“进程”， bash_command="python process.py {{ ti.xcom_pull(task_ids='extract') }}", ） **XCom 与 EmailOperator:** 从airflow.sdk导入任务从airflow.providers.smtp.operators.smtp导入EmailOperator @任务 def get_ip(): 返回“192.168.1.1” @任务（多个输出=真） def compose_email(external_ip): 返回{ 'subject': f'从 {external_ip} 连接的服务器', 'body': f'您的服务器是从 {external_ip}
连接的' } 电子邮件信息 = 撰写电子邮件(get_ip()) 电子邮件操作员( task_id='发送电子邮件', 至='example@example.com', 主题=email_info['主题'], html_content=email_info['body'] ） ## 动态工作流程根据运行时条件或外部数据动态创建任务。 ### 带循环的动态任务生成从airflow.sdk导入DAG 从airflow.operators.empty导入EmptyOperator 使用 DAG("dynamic_loop_example", ...) 作为 dag：开始 = EmptyOperator(task_id="开始") 结束 = EmptyOperator(task_id="结束") # 动态创建任务选项= [“branch_a”，“branch_b”，“branch_c”，“branch_d”] 对于选项中的选项：任务 = EmptyOperator(task_id=选项) 开始>>任务>>结束 ### 动态任务映射映射任务输出以创建动态并行任务：从airflow.decorators导入任务 @任务 def 提取（）： # 返回要处理的项目列表返回 [1, 2, 3, 4, 5] @任务 def 变换（项目）： # 处理单个项目退货*2 @任务默认负载（项目）： # 接收所有已转换的项目 print(f"已加载 {len(items)} 项：{items}") # 动态映射数据=提取（） Transformed = Transform.expand(item=data) # 创建 5 个并行任务负载（转换后） **与经典运算符的映射：** 从airflow.operators.bash导入BashOperator 类提取操作符（基础操作符）： def 执行（自身，上下文）：返回[“文件1.csv”，“文件2.csv”，“文件3.csv”] 类 TransformOperator(BaseOperator): def __init__(自身，输入，**kwargs)： super().__init__(**kwargs) self.input = 输入 def 执行（自身，上下文）： # 处理单个文件返回 f“已处理_{self.input}” 提取 = ExtractOperator(task_id="提取") 变换 = TransformOperator.partial(task_id="transform").expand(input=extract.output) ### 任务组映射映射整个任务组：从airflow.decorators导入任务，task_group @任务 def add_one(值): 返回值+1 @任务 def 双精度（值）：返回值*2 @任务组 def process_group(值): 递增 = add_one(值) 返回双精度（递增） @任务 def 聚合（结果）： print(f"结果：{结果}") # 将任务组映射到值上结果 = process_group.expand(value=[1, 2, 3, 4, 5]) 合计（结果） ### 带映射的部分参数混合静态和动态参数： @任务 def 进程（基本路径，文件名）：完整路径 = f"{基本路径}/{文件名}" 返回f“已处理的{full_path}” # 静态参数'base_path'，动态'文件名' 结果= process.partial(base_path="/data").expand( 文件名=[“文件1.csv”，“文件2.csv”，“文件3.csv”] ） ## 任务流 API 使用自动 XCom 处理和更简洁的语法来编写 Airflow DAG 的现代方法。 ### 基本任务流程示例从airflow.decorators导入dag，任务进口摆锤 @达格（ dag_id="taskflow_example", start_date=pendulum.datetime(2023, 10, 26, tz="UTC"), 时间表=无，追赶=假，） def my_taskflow_dag(): @任务 def 提取（）： data_string = '{“1001”：301.27，“1002”：433.21，“1003”：502.22}' 导入 json 返回 json.loads(data_string) @任务 def 变换（order_data_dict）：总计 = sum(order_data_dict.values()) 返回{“total_order_value”：总计} @任务定义负载（摘要）： print(f"订单总价值：{summary['total_order_value']:.2f}") # 函数调用自动创建任务依赖关系订单数据=提取（）摘要 = 变换（订单数据）负载（摘要） # 实例化 DAG my_taskflow_dag() ### 多个输出从任务返回多个值： @任务（多个输出=真） def extract_data(): 返回{ “订单”：[1,2,3], '客户'：['A'，'B'，'C']， “收入”：1000.50 } @任务 def process_orders(订单): print(f"正在处理 {len(orders)} 订单") @任务 def process_customers（客户）： print(f"正在处理 {len(customers)} 客户") # 访问各个输出数据=提取数据（） process_orders(数据['订单']) process_customers(数据['客户']) ### 将任务流与传统运算符混合从airflow.decorators导入dag，任务从airflow.providers.standard.operators.bash导入BashOperator 进口摆锤 @达格（ start_date=pendulum.datetime(2023, 1, 1, tz="UTC"), 时间表=无，） def mix_dag(): @任务 def get_date(): 从日期时间导入日期时间 return datetime.now().strftime("%Y%m%d") # 传统运算符 bash_task = BashOperator( 任务ID =“打印日期”， bash_command="echo 处理 {{ ti.xcom_pull(task_ids='get_date') }} 的数据" ） @任务 def process_results(): print("处理完成") # 混合任务类型日期 = get_date() 日期 >> bash_task >> process_results() 混合达格() ### 任务的虚拟环境隔离任务依赖关系：从airflow.decorators导入dag，任务进口摆锤 @达格（ dag_id="virtualenv_example", start_date=pendulum.datetime(2023, 10, 26, tz="UTC"), ） def virtualenv_dag(): @task.virtualenv( 要求=["pandas==1.5.0","numpy==1.23.0"], system_site_packages=False ） defanalyze_data（）：将 pandas 导入为 pd 将 numpy 导入为 np df = pd.DataFrame({'col1': [1, 2, 3], 'col2': [4, 5, 6]}) 结果 = np.mean(df['col1']) 返回浮点数（结果） @任务 def 报告结果（平均值）： print(f"平均值：{mean_value}") 结果=分析数据() 报告结果（结果） virtualenv_dag() ## 基于资产的调度根据数据资产（以前称为数据集）而不是时间来安排 DAG。 ### 生产者-消费者模式从airflow.sdk导入DAG，资产从airflow.operators.bash导入BashOperator 从日期时间导入日期时间 # 定义资产 customer_data = 资产("s3://my-bucket/customers.parquet") # 生产者 DAG 与 DAG( dag_id =“生产者_dag”，开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”，）作为制作人： Bash 运算符( task_id =“生成数据”， bash_command =“python生成_customers.py”， outlets=[customer_data] # 将资产标记为已更新） # Consumer DAG - 资产更新时触发与 DAG( dag_id =“consumer_dag”， Schedule=[customer_data], # 由资产触发开始日期 = 日期时间(2023, 1, 1), 追赶=假，）作为消费者： Bash 运算符( task_id =“进程数据”， bash_command =“python process_customers.py” ） ### 多重资产依赖 **AND 逻辑（所有资产必须更新）：** 从airflow.datasets导入数据集 asset_1 = 数据集("s3://bucket/file1.csv") asset_2 = 数据集("s3://bucket/file2.csv") 与 DAG( dag_id="wait_for_both", Schedule=[asset_1 & asset_2], # 两者都必须更新开始日期 = 日期时间(2023, 1, 1), ）：通过 **或逻辑（任何资产更新触发）：** asset_1 = 数据集("s3://bucket/file1.csv") asset_2 = 数据集("s3://bucket/file2.csv") 与 DAG( dag_id="triggered_by_either", 时间表=[asset_1 | asset_2], # 两者都可以触发开始日期 = 日期时间(2023, 1, 1), ）：通过 **复杂逻辑：** asset_1 = 数据集("s3://bucket/file1.csv") asset_2 = 数据集("s3://bucket/file2.csv") asset_3 = 数据集("s3://bucket/file3.csv") 与 DAG( dag_id="complex_condition", 时间表=(asset_1 | (asset_2 & asset_3)), # asset_1 OR (asset_2 AND asset_3) 开始日期 = 日期时间(2023, 1, 1), ）：通过 ### 资产别名使用别名进行灵活的资产引用：从airflow.datasets导入数据集，AssetAlias 从airflow.decorators导入任务 # 具有别名的生产者与 DAG(dag_id="alias_ Producer", start_date=datetime(2023, 1, 1)): @task(outlets=[AssetAlias("my-alias")]) def Produce_data(*,outlet_events): # 动态添加实际资产 outlet_events[AssetAlias("my-alias")].add( 数据集（“s3://bucket/my-file.csv”）） # 消费者取决于别名与 DAG( dag_id="alias_consumer", 时间表=AssetAlias(“我的别名”), 开始日期 = 日期时间(2023, 1, 1), ）：通过 ### 访问资产事件信息 @任务 def process_asset_data（*，triggering_asset_events）：对于triggering_asset_events中的事件： print(f"资产：{event.asset.uri}") print(f"时间戳：{event.timestamp}") print(f"额外：{event.extra}") ## 调度模式 ### Cron 表达式 #每天午夜时间表=“0 0 * * *” #每周一上午9点时间表=“0 9 * * 1” # 每 15 分钟一次时间表=“*/15 * * * *” # 每月第一天中午时间表=“0 12 1 * *” # 工作日下午 6 点时间表=“0 18 * * 1-5” ### 时间增量调度从日期时间导入时间增量与 DAG( dag_id="timedelta_schedule", 开始日期 = 日期时间(2023, 1, 1), Schedule=timedelta(hours=6), # 每 6 小时一次）：通过 ### 预设时间表 # 常用预设 Schedule="@once" # 运行一次 Schedule="@hourly" # 每小时 Schedule="@daily" # 每天午夜 Schedule="@weekly" # 每个周日午夜 Schedule="@monthly" # 每月第一天午夜 Schedule="@yearly" # 1 月 1 日午夜 Schedule=None # 仅手动触发 ### 追赶和回填 **追赶：** 与 DAG( dag_id="catchup_example", 开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”， catchup=True, # 运行所有错过的时间间隔）：通过与 DAG( dag_id="no_catchup", 开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”， catchup=False, # 只运行最新的间隔）：通过 **手动回填：** # 回填特定日期范围气流DAG回填\ --开始日期 2023-01-01 \ --结束日期 2023-01-31 \ 我的_dag_id # 回填标记成功（不执行）气流DAG回填\ --开始日期 2023-01-01 \ --结束日期 2023-01-31 \ --标记成功 \ 我的_dag_id ## 生产模式 ### 错误处理和重试 **任务级重试：** 从airflow.operators.bash导入BashOperator 从日期时间导入时间增量带重试的任务 = BashOperator( task_id =“重试任务”， bash_command =“python might_fail.py”，重试=3， retry_delay=timedelta(分钟=5), retry_exponential_backoff=真， max_retry_delay=timedelta(分钟=30), ） **DAG 级别默认参数：** 从日期时间导入日期时间，时间增量默认参数 = { '所有者'：'数据团队'， 'depends_on_past'：错误， '电子邮件': ['alerts@company.com'], 'email_on_failure'：正确， 'email_on_retry'：错误， “重试”：2， 'retry_delay': timedelta(分钟=5), } 与 DAG( dag_id =“生产_dag”，默认参数=默认参数，开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”，）：通过 ### 任务并发控制 **每任务并发：** 从airflow.operators.bash导入BashOperator 从日期时间导入时间增量 # 限制该任务的并发实例数有限任务 = BashOperator( task_id="有限任务", bash_command="echo '处理'", max_active_tis_per_dag=3 # 最多运行 3 个实例） **DAG 级并发：** 与 DAG( dag_id =“并发_dag”，开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”， max_active_runs=5, # 最多同时运行 5 个 DAG concurrency=10, # 所有运行最多 10 个任务实例）：通过 ### 幂等性使任务幂等以便安全重试： @任务 def idempote_load(**上下文): 执行日期 = 上下文['ds'] # 首先删除该日期的现有数据删除查询 = f""" 从目标表中删除 WHERE 日期 = '{execution_date}' ”“” 执行_sql（删除_查询） # 插入新数据插入查询 = f""" 插入目标表从源中选择 * WHERE 日期 = '{execution_date}' ”“” 执行_sql（插入_查询） ### SLA 和警报从日期时间导入时间增量 def sla_miss_callback（dag，task_list，blocking_task_list，slas，blocking_tis）： print(f"{task_list} 缺少 SLA") # 向监控系统发送警报与 DAG( dag_id =“sla_dag”，开始日期 = 日期时间(2023, 1, 1), 时间表=“@每日”，默认参数={ 'sla': timedelta(hours=2), # 任务应在 2 小时内完成 }, sla_miss_callback=sla_miss_callback, ）：通过 ### 任务回调 def on_failure_callback（上下文）： print(f"任务 {context['task_instance'].task_id} 失败") # 发送到 Slack、PagerDuty 等。 def on_success_callback(上下文): print(f"任务 {context['task_instance'].task_id} 成功") def on_retry_callback(上下文): print(f"任务 {context['task_instance'].task_id} 正在重试") task_with_callbacks = BashOperator( task_id =“受监控的任务”， bash_command =“python my_script.py”， on_failure_callback=on_failure_callback, on_success_callback=on_success_callback, on_retry_callback=on_retry_callback, ） ### Docker 部署 **用于本地开发的 Docker Compose：** 版本：'3' 服务： postgres：图片：postgres：13 环境： POSTGRES_USER：气流 POSTGRES_PASSWORD：气流 POSTGRES_DB：气流网络服务器：图片：apache/airflow:2.7.0 取决于： - postgres 环境： AIRFLOW__CORE__EXECUTOR：LocalExecutor AIRFLOW__DATABASE__SQL_ALCHEMY_CONN：postgresql+psycopg2://airflow:airflow@postgres/airflow 端口： - “8080:8080” 命令：网络服务器调度程序：图片：apache/airflow:2.7.0 取决于： - postgres 环境： AIRFLOW__CORE__EXECUTOR：LocalExecutor AIRFLOW__DATABASE__SQL_ALCHEMY_CONN：postgresql+psycopg2://airflow:airflow@postgres/airflow 命令：调度程序 ### Kubernetes 执行器 **Kubernetes执行器配置：** # 在airflow.cfg中 [Kubernetes] 命名空间=气流 worker_container_repository = 我的注册表/气流工人容器标签 = 2.7.0 删除_worker_pods = True delete_worker_pods_on_failure = False [核心] 执行器 = KubernetesExecutor **特定任务的 Pod 覆盖：** 从 kubernetes.client 将模型导入为 k8s 带 GPU 的任务 = BashOperator( task_id =“gpu_task”， bash_command =“python train_model.py”，执行器配置={ “pod_override”：k8s.V1Pod（规格=k8s.V1PodSpec( 容器=[ k8s.V1Container( 名称=“基础”，资源=k8s.V1ResourceRequirements( 限制={"nvidia.com/gpu": "1"} ）） ] ）） } ） ### 监控和日志记录 **结构化日志记录：** 从airflow.decorators导入任务导入日志记录 @任务 def 受监控的任务（）：记录器=logging.getLogger(__name__) logger.info("开始数据处理", extra={ 'process_id': 'abc123', “记录数”：1000 }) 尝试：处理数据（） logger.info("处理完成") 除了异常 e： logger.error(f"处理失败：{str(e)}", extra={ 'error_type': 类型(e).__name__ }) 提高 **StatsD 指标：** 从airflow.stats导入统计数据 @任务 def task_with_metrics(): Stats.incr('my_dag.task_started') 开始时间 = 时间.time() 处理数据（）持续时间 = time.time() - 开始时间 Stats.timing('my_dag.task_duration', 持续时间) Stats.incr('my_dag.task_completed') ## 最佳实践 ### DAG 设计 1. **保持 DAG 简单**：将复杂的工作流程分解为多个 DAG 2. **使用描述性名称**：dag_id和task_id应该是不言自明的 3. **幂等任务**：任务重新运行时应产生相同的结果 4. **小型 XCom** ：将 XCom 数据保持在 1MB 以下 5. **外部存储** ：使用S3/GCS进行大数据，传递引用 6. **正确的依赖关系**：建模真实的依赖关系，避免不必要的依赖关系 7. **错误处理**：使用重试、回调和正确的错误日志记录 8. **资源管理**：设置适当的任务并发限制 ### 代码组织达格斯/ ├── 常见/ │ ├── __init__.py │ ├──operators.py # 自定义运算符 │ ├──sensors.py # 自定义传感器 │ └── utils.py # 实用函数 ├── 等/ │ ├── customer_pipeline.py │ ├── order_pipeline.py │ └── 产品管道.py ├── 毫升/ │ ├── 训练_dag.py │ └── inference_dag.py └── 保养/ ├── cleanup_dag.py └── backup_dag.py ### 测试 DAG **单元测试：** 导入pytest 从airflow.models导入DagBag def test_dag_loaded(): dagbag = DagBag(dag_folder='dags/', include_examples=False) 断言 len(dagbag.import_errors) == 0 def test_task_count(): dagbag = DagBag(dag_folder='dags/') dag = dagbag.get_dag('my_dag') 断言 len(dag.tasks) == 5 def test_task_dependency(): dagbag = DagBag(dag_folder='dags/') dag = dagbag.get_dag('my_dag') 提取 = dag.get_task('提取') 变换 = dag.get_task('变换') 在 extract.downstream_list 中断言转换 **集成测试：** 从airflow.models导入DagBag 从 airflow.utils.state 导入状态 def test_dag_runs(): dagbag = DagBag(dag_folder='dags/') dag = dagbag.get_dag('my_dag') # 测试 DAG 运行 dag_run = dag.create_dagrun( 状态=状态.RUNNING, 执行日期 = 日期时间(2023, 1, 1), run_type='手动' ） # 运行特定任务 task_instance = dag_run.get_task_instance('提取') 任务实例.run() 断言task_instance.state == State.SUCCESS ### 性能优化 1. **使用可延迟运算符**：用于传感器和长时间运行的等待 2. **动态任务映射**：用于并行处理 3. **合适的执行器** ：根据规模选择（Local、Celery、Kubernetes） 4. **连接池**：重用数据库连接 5. **任务并行**：适当设置 max_active_runs 和并发数 6. **延迟加载**：在 DAG 解析时不要执行繁重的逻辑 7. **外部存储**：保持元数据数据库轻量 ### 安全 1. **Secrets Management** ：使用 Airflow Secrets 后端（未硬编码） 2. **连接加密** ：对数据库使用加密连接 3. **RBAC**：启用基于角色的访问控制 4. **审核日志记录**：启用审核日志以确保合规性 5. **网络隔离**：限制worker网络访问 6. **凭证轮换**：定期轮换凭证 ### 配置管理 # 使用变量进行配置从airflow.models导入变量 config = Variable.get("my_config", deserialize_json=True) api_key = Variable.get("api_key") # 使用连接进行外部服务从airflow.hooks.base导入BaseHook conn = BaseHook.get_connection('my_postgres') db_url = f"postgresql://{conn.login}:{conn.password}@{conn.host}:{conn.port}/{conn.schema}" ## 常见模式和示例请参阅 EXAMPLES.md 了解 18 个以上详细的实际示例，包括： * ETL管道 * 机器学习工作流程 * 数据质量检查 * 多云编排 * 事件驱动架构 * 复杂的分支逻辑 * 动态任务生成 * 基于资产的调度 * 传感器模式 * 错误处理策略 ## 故障排除 ### DAG 未出现在 UI 中 1.检查DAG文件中的Python语法错误 2.验证DAG文件是否位于正确的目录中 3. 检查dag_id是否唯一 4. 如果您希望计划运行，请确保计划不是“无” 5. 检查调度程序日志是否存在导入错误 ### 任务未运行 1.检查任务依赖关系是否正确 2. 验证上游任务是否成功 3. 检查任务并发限制 4. 确保执行器有可用的槽位 5. 检查任务日志是否有错误 ### 性能问题 1.降低DAG复杂度（分成多个DAG） 2.优化任务中的SQL查询 3. 根据规模使用适当的执行器 4.启用任务并行 5. 检查慢速传感器（使用可延迟模式） 6. 监控元数据数据库性能 ### 常见错误 **导入错误：** # 不好 - DAG 级别的导入解析缓慢从 Heavy_library 导入过程与 DAG(...): 通过 # 好 - 导入任务内部与 DAG(...): @任务 def my_task(): 从 Heavy_library 导入过程进程（） **循环依赖：** # 这会失败任务1 >> 任务2 >> 任务3 >> 任务1 # 循环！ # 必须是非循环的任务1 >> 任务2 >> 任务3 **大型 XCom：** # 不好 - 在 XCom 中存储大量数据 @任务定义过程（）： large_df = pd.read_csv('big_file.csv') return large_df # 太大了！ # 好 - 商店参考 @任务定义过程（）： large_df = pd.read_csv('big_file.csv') 路径 = save_to_s3(large_df) 返回路径 # 只是路径 ## 资源 * **官方文档**： * **Airflow GitHub** ： * **天文学家指南**： * **社区 Slack** ： * **堆栈溢出**：标记`apache-airflow` * **很棒的气流**： ** ** **技能版本**：1.0.0 **最后更新**：2025 年 1 月 **Apache Airflow 版本**：2.7+ **技能类别**：数据工程、工作流编排、管道管理

Apache 气流编排

包含技能