跳到主要内容

Airflow SubDagOperator 详解

介绍​

在 Apache Airflow 中，SubDagOperator 是一种特殊的操作符，用于将一组任务封装为一个子工作流（SubDAG）。通过使用 SubDagOperator，你可以将复杂的工作流分解为更小、更易管理的部分，从而提高代码的可读性和可维护性。

备注

SubDagOperator 在 Airflow 2.0 中已被标记为弃用，推荐使用 TaskGroup 来替代。但对于初学者来说，了解 SubDagOperator 仍然是有价值的，因为它可以帮助你理解如何组织和简化复杂的工作流。

什么是 SubDagOperator？​

SubDagOperator 允许你将一组任务封装为一个子工作流（SubDAG），并将其作为一个单独的任务插入到主 DAG 中。子工作流可以有自己的任务和依赖关系，但它们仍然属于主 DAG 的一部分。

主要特点​

  • 封装性：将一组任务封装为一个子工作流，简化主 DAG 的结构。
  • 可重用性：可以在多个 DAG 中重复使用相同的子工作流。
  • 独立性：子工作流可以有自己的任务和依赖关系，独立于主 DAG。

如何使用 SubDagOperator​

基本用法​

要使用 SubDagOperator，首先需要定义一个子工作流函数，然后将其传递给 SubDagOperator。以下是一个简单的示例：

python
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.subdag_operator import SubDagOperator
from airflow.utils.dates import days_ago

def create_subdag(parent_dag_name, child_dag_name, args):
    subdag = DAG(
        dag_id=f'{parent_dag_name}.{child_dag_name}',
        default_args=args,
        schedule_interval=None,
    )

    with subdag:
        start = DummyOperator(task_id='start')
        end = DummyOperator(task_id='end')
        start >> end

    return subdag

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

with DAG('main_dag', default_args=default_args, schedule_interval=None) as dag:
    start = DummyOperator(task_id='start')
    subdag = SubDagOperator(
        task_id='subdag',
        subdag=create_subdag('main_dag', 'subdag', default_args),
    )
    end = DummyOperator(task_id='end')

    start >> subdag >> end

在这个示例中，create_subdag 函数定义了一个简单的子工作流，包含两个 DummyOperator 任务。然后，SubDagOperator 将这个子工作流插入到主 DAG 中。

输入和输出​

  • 输入：SubDagOperator 需要传入一个子工作流函数，该函数返回一个 DAG 对象。
  • 输出：SubDagOperator 会将子工作流作为一个任务插入到主 DAG 中。

实际案例​

假设你正在处理一个复杂的数据处理工作流，其中包含多个阶段的数据清洗、转换和加载。你可以使用 SubDagOperator 将每个阶段封装为一个子工作流，从而简化主 DAG 的结构。

python
def create_data_cleaning_subdag(parent_dag_name, child_dag_name, args):
    subdag = DAG(
        dag_id=f'{parent_dag_name}.{child_dag_name}',
        default_args=args,
        schedule_interval=None,
    )

    with subdag:
        clean_data_1 = DummyOperator(task_id='clean_data_1')
        clean_data_2 = DummyOperator(task_id='clean_data_2')
        clean_data_1 >> clean_data_2

    return subdag

def create_data_transformation_subdag(parent_dag_name, child_dag_name, args):
    subdag = DAG(
        dag_id=f'{parent_dag_name}.{child_dag_name}',
        default_args=args,
        schedule_interval=None,
    )

    with subdag:
        transform_data_1 = DummyOperator(task_id='transform_data_1')
        transform_data_2 = DummyOperator(task_id='transform_data_2')
        transform_data_1 >> transform_data_2

    return subdag

with DAG('data_pipeline', default_args=default_args, schedule_interval=None) as dag:
    start = DummyOperator(task_id='start')
    data_cleaning = SubDagOperator(
        task_id='data_cleaning',
        subdag=create_data_cleaning_subdag('data_pipeline', 'data_cleaning', default_args),
    )
    data_transformation = SubDagOperator(
        task_id='data_transformation',
        subdag=create_data_transformation_subdag('data_pipeline', 'data_transformation', default_args),
    )
    end = DummyOperator(task_id='end')

    start >> data_cleaning >> data_transformation >> end

在这个案例中，我们将数据清洗和数据转换分别封装为两个子工作流，并使用 SubDagOperator 将它们插入到主 DAG 中。

总结​

SubDagOperator 是一个强大的工具，可以帮助你将复杂的工作流分解为更小、更易管理的部分。虽然它在 Airflow 2.0 中已被弃用，但对于初学者来说，了解它的工作原理仍然是有价值的。

提示

如果你正在使用 Airflow 2.0 或更高版本，建议使用 TaskGroup 来替代 SubDagOperator，以获得更好的性能和更简洁的代码结构。

附加资源​

练习​

  1. 创建一个包含多个子工作流的 DAG，每个子工作流包含至少两个任务。
  2. 尝试将 SubDagOperator 替换为 TaskGroup，并比较两者的代码结构和性能。