气流:将动态值传递给Sub DAG操作员

Question

我是Airflow的新手.
我遇到过一个场景,其中Parent DAG需要将一些动态数字(比方说n)传递给Sub DAG.
SubDAG将使用此数字动态创建n并行任务.

气流文档未涵盖实现此目的的方法.所以我探索了几种方法:

选项-1(使用xcom Pull)

我试图传递为xcom值,但由于某种原因,SubDAG没有解析为传递的值.

父Dag文件

def load_dag(**kwargs):
    number_of_runs = json.dumps(kwargs['dag_run'].conf['number_of_runs'])
    dag_data = json.dumps({
        "number_of_runs": number_of_runs
    })
    return dag_data

# ------------------ Tasks ------------------------------
load_config = PythonOperator(
    task_id='load_config',
    provide_context=True,
    python_callable=load_dag,
    dag=dag)


t1 = SubDagOperator(
    task_id=CHILD_DAG_NAME,
    subdag=sub_dag(PARENT_DAG_NAME, CHILD_DAG_NAME, default_args, "'{{ ti.xcom_pull(task_ids='load_config') }}'" ),
    default_args=default_args,
    dag=dag,
)

Sub Dag文件

def sub_dag(parent_dag_name, child_dag_name, args, num_of_runs):
    dag_subdag = DAG(
        dag_id='%s.%s' % (parent_dag_name, child_dag_name),
        default_args=args,
        schedule_interval=None)

    variabe_names = {}

    for i in range(num_of_runs):
        variabe_names['task' + str(i + 1)] =  DummyOperator(
        task_id='dummy_task',
        dag=dag_subdag,
    )

    return dag_subdag

选项 - 2

我也试图传递number_of_runs一个全局变量,这是无效的.

选项 - 3

我们还尝试将此值写入数据文件.但是子DAG正在投掷File doesn't exist error.这可能是因为我们正在动态生成此文件.

有人可以帮我弄这个吗.

Answer 1

我已经使用选项 3 完成了此操作。关键是如果文件不存在，则返回没有任务的有效 dag。因此，load_config 将生成一个文件，其中包含您的任务数量或更多信息（如果需要）。你的 subdag 工厂看起来像这样：

def subdag(...):
    sdag = DAG('%s.%s' % (parent, child), default_args=args, schedule_interval=timedelta(hours=1))
    file_path = "/path/to/generated/file"
    if os.path.exists(file_path):
        data_file = open(file_path)
        list_tasks = data_file.readlines()
        for task in list_tasks:
            DummyOperator(
                  task_id='task_'+task,
                  default_args=args,
                  dag=sdag,
            )
    return sdag

在 dag 生成时，您将看到一个没有任务的 subdag。在dag执行时，load_config完成后，您可以看到动态生成的subdag