如何为动态计算列表中的每个项执行Airflow DockerOperator？

需求：基于@task生成的动态计算列表，为每个列表项运行Docker镜像，并将该项作为环境变量传入容器。

现有示例DAG可基于SQL查询结果的列表，通过.expand()方法批量执行打印任务，但需要实现对DockerOperator的类似批量逻辑，期望的伪代码如下：

for single_item in list_fetched_from_task:
    DockerOperator(
        task_id='run-some-image-for-single-item',
        image='some-image:latest',
        docker_url='unix:///var/run/docker.sock',
        network_mode='host',
        environment={
            "SINGLE_ITEM": single_item
        },
        dag=dag
    )

附原有打印任务示例代码：

from airflow import DAG
from airflow.decorators import task
from datetime import datetime
from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook


@task
def get_all_items():
    mssqlServer = MsSqlHook(mssql_conn_id="my_mssql")
    raw_item_names = mssqlServer.get_records(
        """
        use database_name;
        select itemName from database_name.fds.items;
        """
    )

    item_names = []
    for item_name in raw_item_names:
        item_names.append(item_name[0])
    
    return item_names # ["Item 1", "Item 2", "Item 3"]

@task
def do_thing_for_one_item(single_item):
    print("Doing thing for item: " + single_item)


with DAG(dag_id="do-thing-for-each-item-process", start_date=datetime(2022, 10, 29)) as dag:
    do_thing_for_one_item.expand(single_item=get_all_items())

解决方案 #

Airflow 2.x的**动态任务映射（Dynamic Task Mapping）**支持传统Operator（包括DockerOperator）直接使用.expand()实现批量执行，无需手动循环创建任务。

完整实现代码 #

from airflow import DAG
from airflow.decorators import task
from datetime import datetime
from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
from airflow.providers.docker.operators.docker import DockerOperator


@task
def get_all_items():
    mssqlServer = MsSqlHook(mssql_conn_id="my_mssql")
    raw_item_names = mssqlServer.get_records(
        """
        use database_name;
        select itemName from database_name.fds.items;
        """
    )

    item_names = []
    for item_name in raw_item_names:
        item_names.append(item_name[0])
    
    return item_names # ["Item 1", "Item 2", "Item 3"]


with DAG(dag_id="run-docker-for-each-item", start_date=datetime(2022, 10, 29), catchup=False) as dag:
    # 定义DockerOperator的公共参数，用partial固定不变的配置
    docker_task = DockerOperator.partial(
        task_id="run_docker_for_item",
        image="some-image:latest",
        docker_url="unix:///var/run/docker.sock",
        network_mode="host",
        # 可选：容器运行后自动删除
        auto_remove=True
    ).expand(
        # 动态传递environment参数，每个列表项对应一个环境变量配置
        environment=[{"SINGLE_ITEM": item} for item in get_all_items()]
    )

关键说明 #

• 使用DockerOperator.partial()定义所有固定不变的参数（如镜像地址、Docker套接字路径等），避免重复代码。
• 通过.expand(environment=...)传递动态参数：将get_all_items()返回的列表转换为包含环境变量字典的列表，每个字典对应一个列表项。
• Airflow会自动为每个动态生成的任务创建唯一的task_id（例如run_docker_for_item__Item_1），避免任务ID冲突。
• 容器启动时，SINGLE_ITEM环境变量会被设置为对应的列表项值，容器内的应用可以直接读取该变量。

内容的提问来源于stack exchange，提问作者Antoine Dahan