Airflow 2.2.3中Task与TaskGroup间如何实现XCom Pull取值？

问题根源 #

你当前的代码核心问题是在DAG静态解析阶段尝试获取运行时才生成的XCom数据。Airflow调度器解析DAG文件时，task1.output或模板字符串{{ ti.xcom_pull(...) }}并不会返回实际的文件列表——XCom是任务执行时才会产生的结果，此时循环遍历的是Airflow的XComArg对象或字符串本身，根本无法生成对应每个文件的处理任务。

可行解决方案 #

方案1：升级到Airflow 2.3+ 使用Dynamic Task Mapping（推荐） #

如果可以升级Airflow版本到2.3及以上，Dynamic Task Mapping是最优雅的方案，它支持在运行时根据上游任务输出动态生成任务实例，无需预先定义所有任务。

修改后的代码示例（使用TaskFlow API）：

from airflow.decorators import dag, task
from datetime import datetime

def process_file(file_path):
    print(f"Processing file: {file_path}")
    # 替换为实际文件处理逻辑

default_args = {
    'start_date': datetime(2022, 1, 1)
}

@dag('process_files', default_args=default_args, schedule_interval='@daily')
def process_files_dag():
    @task
    def read_files_from_table():
        # 实际从SFTP获取文件列表
        files = ['file1.txt', 'file2.csv', 'file3.json', 'x_file.txt']
        # 提前过滤不需要处理的文件
        return [f for f in files if 'x_' not in f]

    @task
    def process_file_task(file_path):
        process_file(file_path)

    # 动态映射生成对应每个文件的任务
    file_list = read_files_from_table()
    process_file_task.expand(file_path=file_list)

dag = process_files_dag()

方案2：Airflow 2.2.3 兼容方案（单任务内并行处理） #

如果无法升级版本，可在单个PythonOperator中使用线程/进程池实现文件并行处理，缺点是无法单独监控每个文件的处理状态，但能满足并行需求：

from airflow.operators.python import PythonOperator
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

def read_files_from_table():
    # 实际从SFTP获取文件列表
    files = ['file1.txt', 'file2.csv', 'file3.json', 'x_file.txt']
    return [f for f in files if 'x_' not in f]

def process_file(file_path):
    print(f"Processing file: {file_path}")
    # 替换为实际文件处理逻辑

def process_files_in_parallel(**context):
    file_list = context['ti'].xcom_pull(task_ids='read_files_from_table')
    # 线程池并行处理，可根据资源调整max_workers
    with ThreadPoolExecutor(max_workers=3) as executor:
        executor.map(process_file, file_list)

default_args = {
    'start_date': datetime(2022, 1, 1)
}

with DAG('process_files', default_args=default_args, schedule_interval='@daily') as dag:
    task_get_files = PythonOperator(
        task_id='read_files_from_table',
        python_callable=read_files_from_table
    )

    task_process_all = PythonOperator(
        task_id='process_all_files',
        python_callable=process_files_in_parallel,
        provide_context=True
    )

    task_get_files >> task_process_all

方案3：Airflow 2.2.3 兼容方案（触发子DAG） #

若需要单独监控每个文件的处理状态，可使用TriggerDagRunOperator触发子DAG处理单个文件，但需要预先定义子DAG：

子DAG定义（sub_dag_process_file.py）

from airflow.operators.python import PythonOperator
from airflow.models import DAG
from datetime import datetime

def process_file(file_path):
    print(f"Processing file: {file_path}")
    # 替换为实际文件处理逻辑

def create_subdag(parent_dag_name, child_dag_name, args):
    with DAG(
        f"{parent_dag_name}.{child_dag_name}",
        start_date=args['start_date'],
        schedule_interval=None,
        default_args=args
    ) as subdag:
        PythonOperator(
            task_id='process_single_file',
            python_callable=process_file,
            op_kwargs={'file_path': '{{ dag_run.conf["file_path"] }}'}
        )
    return subdag

主DAG定义

from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime
from sub_dag_process_file import create_subdag

def read_files_from_table(**context):
    # 实际从SFTP获取文件列表
    files = ['file1.txt', 'file2.csv', 'file3.json', 'x_file.txt']
    filtered_files = [f for f in files if 'x_' not in f]
    context['ti'].xcom_push(key='file_list', value=filtered_files)
    return filtered_files

default_args = {
    'start_date': datetime(2022, 1, 1)
}

with DAG('process_files', default_args=default_args, schedule_interval='@daily') as dag:
    task_get_files = PythonOperator(
        task_id='read_files_from_table',
        python_callable=read_files_from_table,
        provide_context=True
    )

    with TaskGroup('trigger_subdags') as trigger_group:
        # 注意：此方式需提前预估可能的文件数量或在DAG解析时能获取到文件列表（仅适用于静态列表场景）
        # 若文件列表完全动态，此方案无法在DAG解析阶段生成TriggerDagRunOperator，需结合方案2或升级版本
        file_list = ['file1.txt', 'file2.csv', 'file3.json']  # 仅示例，静态场景可用
        for file in file_list:
            TriggerDagRunOperator(
                task_id=f'trigger_process_{file}',
                trigger_dag_id='process_files.process_single_file',
                conf={'file_path': file}
            )

    task_get_files >> trigger_group

注意：此方案仅适用于文件列表可在DAG解析阶段获取的场景，若文件列表完全动态（比如每次运行从SFTP获取不同列表），此方案无法直接使用，推荐优先考虑方案1或方案2。

内容的提问来源于stack exchange，提问作者Kumar Padhy