Airflow 2.1.2中如何在PythonOperator可执行函数内设置inlets和outlets

在Airflow 2.1.2中，你尝试在PythonOperator的可执行函数内动态设置inlets和outlets的方式无法生效，原因是：

• inlets和outlets是Airflow Task对象的静态定义属性，会在DAG加载解析阶段被读取（比如DataHub的集成插件会在此阶段扫描这些属性生成血缘关系）。
• 运行时通过kwargs.get("task")获取的Task对象是副本，修改它的属性不会同步到DAG定义的原始对象；而TaskInstance（即ti）本身并不具备inlets/outlets属性，修改自然无效。

可行的实现方式 #

1. 静态定义（数据集固定场景）

直接在PythonOperator初始化时指定inlets和outlets：

from datahub_provider import entities

def executable_func(**kwargs):
    # 你的业务逻辑
    ...

dag_task = PythonOperator(
    task_id="task_id",
    python_callable=executable_func,
    provide_context=True,
    inlets=[entities.Dataset(source, data_path)],
    outlets=[entities.Dataset(source, data_path)],
    dag=dag,
)

2. DAG构建阶段动态生成（数据集可提前计算场景）

如果数据集需要基于配置、数据库查询等提前计算的逻辑生成，可以在DAG构建时调用函数动态生成：

from datahub_provider import entities

def get_dynamic_datasets():
    # 这里编写动态获取数据集的逻辑，比如读取配置文件或查询数据库
    source = "your_data_source"
    data_path = "dynamic_table_path"
    return [entities.Dataset(source, data_path)]

def executable_func(**kwargs):
    # 你的业务逻辑
    ...

dag_task = PythonOperator(
    task_id="task_id",
    python_callable=executable_func,
    provide_context=True,
    inlets=get_dynamic_datasets(),
    outlets=get_dynamic_datasets(),
    dag=dag,
)

3. 运行时主动上报（依赖运行时结果场景）

如果数据集必须在任务运行时才能确定（比如依赖上游任务的输出结果），则无法通过inlets/outlets静态属性实现，需要直接调用DataHub的API手动上报血缘关系。例如在可执行函数中添加：

from datahub_provider.entities import Dataset
from datahub.metadata.com.linkedin.pegasus2avro.dataset import DatasetLineageType
from datahub.api.entities.dataset import Dataset as DataHubDataset

def executable_func(**kwargs):
    # 你的业务逻辑，获取动态的源数据集和目标数据集
    source_dataset = Dataset("source", "path/to/source")
    target_dataset = Dataset("target", "path/to/target")
    
    # 调用DataHub API上报血缘
    DataHubDataset(urn=f"urn:li:dataset:(urn:li:dataPlatform:{source_dataset.platform},{source_dataset.name},PROD)").add_upstream_lineage(
        upstream_urn=f"urn:li:dataset:(urn:li:dataPlatform:{target_dataset.platform},{target_dataset.name},PROD)",
        lineage_type=DatasetLineageType.TRANSFORMED
    )
    ...

结论 #

标准Airflow 2.1.2安装下，无法通过运行时修改Task对象的inlets/outlets属性来实现动态血缘配置，必须通过上述静态定义、构建阶段动态生成或运行时API上报的方式完成。

内容的提问来源于stack exchange，提问作者Rodion Proskuriakov