如何在Airflow的PythonVirtualenvOperator中获取dag_id与task_id？

由于PythonVirtualenvOperator需要将上下文对象序列化后传递到独立虚拟环境，而dag和task这类对象包含大量无法被dill/pickle序列化的属性，导致无法直接从context中获取。你可以用以下两种方法避开这个问题：

方法一：直接通过op_kwargs传递已知ID #

在定义Operator时，dag_id和task_id都是明确可知的，直接把这两个值作为参数传入函数即可：

修改你的函数，接收dag_id和task_id参数：

def func(dag_id, task_id, **context):
    metric_name = f'dag.{dag_id}.{task_id}.dag_runs'
    # 执行后续逻辑

更新PythonVirtualenvOperator的配置：

task = PythonVirtualenvOperator(
    task_id='task-id',
    python_callable=func,
    system_site_packages=False,
    use_dill=True,
    pip_install_options=pip_install_options,
    op_kwargs={
        'dag_id': dag.dag_id,
        'task_id': 'task-id'  # 这里直接写当前任务的task_id，或用变量维护
    },
    dag=dag,
    provide_context=True  # 若需要其他上下文参数可保留，不需要则可以移除
)

方法二：利用Airflow模板变量传递 #

借助Airflow的模板渲染能力，将dag_id和task_id作为模板参数传入，Operator会在执行前自动替换为实际值：

修改函数：

def func(dag_id, task_id, **context):
    metric_name = f'dag.{dag_id}.{task_id}.dag_runs'
    # 执行后续逻辑

更新Operator配置：

task = PythonVirtualenvOperator(
    task_id='task-id',
    python_callable=func,
    system_site_packages=False,
    use_dill=True,
    pip_install_options=pip_install_options,
    op_kwargs={
        'dag_id': '{{ dag.dag_id }}',
        'task_id': '{{ task.task_id }}'
    },
    dag=dag,
    provide_context=True
)

核心原因说明 #

dag和task对象本身包含数据库连接、调度器引用等复杂属性，无法被序列化传递到虚拟环境，但它们的ID是字符串类型，序列化无压力。以上两种方法都是直接传递ID值，完全避开了序列化复杂对象的问题。

内容的提问来源于stack exchange，提问作者Daniel Watson