OpenLineage 宏¶

包含在 OpenLineage 插件中的宏会集成到 Airflow 的主集合中，并可供使用。

它们可以作为 Jinja 模板调用，例如：

血缘作业和运行宏¶

这些宏

lineage_job_namespace()
lineage_job_name(task_instance)
lineage_run_id(task_instance)

允许将给定 Airflow 任务的运行信息片段注入到发送给远程处理作业的参数中。例如，可以像这样设置 SparkSubmitOperator

SparkSubmitOperator(
    task_id="my_task",
    application="/script.py",
    conf={
        # separated components
        "spark.openlineage.parentJobNamespace": "{{ macros.OpenLineageProviderPlugin.lineage_job_namespace() }}",
        "spark.openlineage.parentJobName": "{{ macros.OpenLineageProviderPlugin.lineage_job_name(task_instance) }}",
        "spark.openlineage.parentRunId": "{{ macros.OpenLineageProviderPlugin.lineage_run_id(task_instance) }}",
    },
)

血缘父级 ID¶

可以使用 linage_parent_id(task_instance) 宏传递相同的信息，但将其压缩为一个字符串

def my_task_function(templates_dict, **kwargs):
    parent_job_namespace, parent_job_name, parent_run_id = templates_dict["parentRun"].split("/")
    ...


PythonOperator(
    task_id="render_template",
    python_callable=my_task_function,
    templates_dict={
        # joined components as one string `<namespace>/<name>/<run_id>`
        "parentRun": "{{ macros.OpenLineageProviderPlugin.lineage_parent_id(task_instance) }}",
    },
    provide_context=False,
    dag=dag,
)