annotating-task-lineage

使用 Inlets 和 Outlets 标注任务血缘

本指南将引导你使用 inlets 和 outlets 为 Airflow 任务添加手动血缘标注。

参考： 如需了解最新支持的运算符和模式，请参阅 OpenLineage provider 开发者指南。

在 Astro 上

使用 inlets 和 outlets 定义的血缘标注会显示在 Astro 增强版的 Lineage 选项卡中，该选项卡提供跨 DAG 和跨部署的血缘视图。这意味着标注会立即显示在 Astro UI 中，让你统一查看整个 Astro 组织的数据流。

何时使用此方法

场景	使用 Inlets/Outlets？
运算符包含 OpenLineage 方法（get_openlineage_facets_on_*）	❌ 直接修改 OL 方法
运算符没有内置的 OpenLineage 提取器	✅ 是
简单的表级血缘已足够	✅ 是
无需自定义代码即可快速设置血缘	✅ 是
需要列级血缘	❌ 使用 OpenLineage 方法或自定义提取器
需要复杂的提取逻辑	❌ 使用 OpenLineage 方法或自定义提取器

注意： Inlets/outlets 是优先级最低的回退方式。如果运算符有对应的 OpenLineage 提取器或方法，则会优先使用。对于没有提取器的运算符，请使用此方法。

---

Inlets/Outlets 支持的类型

你可以使用 OpenLineage Dataset 对象或 Airflow Assets 作为 inlets 和 outlets：

OpenLineage Datasets（推荐）

from openlineage.client.event_v2 import Dataset

# Database tables
source_table = Dataset(
    namespace="postgres://mydb:5432",
    name="public.orders",
)
target_table = Dataset(
    namespace="snowflake://account.snowflakecomputing.com",
    name="staging.orders_clean",
)

# Files
input_file = Dataset(
    namespace="s3://my-bucket",
    name="raw/events/2024-01-01.json",
)

Airflow Assets（Airflow 3+）

from airflow.sdk import Asset

# Using Airflow's native Asset type
orders_asset = Asset(uri="s3://my-bucket/data/orders")

Airflow Datasets（Airflow 2.4+）

from airflow.datasets import Dataset

# Using Airflow's Dataset type (Airflow 2.4-2.x)
orders_dataset = Dataset(uri="s3://my-bucket/data/orders")

---

基本用法

在运算符上设置 Inlets 和 Outlets

from airflow import DAG
from airflow.operators.bash import BashOperator
from openlineage.client.event_v2 import Dataset
import pendulum

# Define your lineage datasets
source_table = Dataset(
    namespace="snowflake://account.snowflakecomputing.com",
    name="raw.orders",
)
target_table = Dataset(
    namespace="snowflake://account.snowflakecomputing.com",
    name="staging.orders_clean",
)
output_file = Dataset(
    namespace="s3://my-bucket",
    name="exports/orders.parquet",
)

with DAG(
    dag_id="etl_with_lineage",
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    schedule="@daily",
) as dag:

    transform = BashOperator(
        task_id="transform_orders",
        bash_command="echo 'transforming...'",
        inlets=[source_table],           # What this task reads
        outlets=[target_table],          # What this task writes
    )

    export = BashOperator(
        task_id="export_to_s3",
        bash_command="echo 'exporting...'",
        inlets=[target_table],           # Reads from previous output
        outlets=[output_file],           # Writes to S3
    )

    transform >> export

多个输入和输出

任务通常会从多个来源读取数据，并写入多个目标位置：

from openlineage.client.event_v2 import Dataset

# Multiple source tables
customers = Dataset(namespace="postgres://crm:5432", name="public.customers")
orders = Dataset(namespace="postgres://sales:5432", name="public.orders")
products = Dataset(namespace="postgres://inventory:5432", name="public.products")

# Multiple output tables
daily_summary = Dataset(namespace="snowflake://account", name="analytics.daily_summary")
customer_metrics = Dataset(namespace="snowflake://account", name="analytics.customer_metrics")

aggregate_task = PythonOperator(
    task_id="build_daily_aggregates",
    python_callable=build_aggregates,
    inlets=[customers, orders, products],      # All inputs
    outlets=[daily_summary, customer_metrics], # All outputs
)

---

在自定义运算符中设置血缘

构建自定义运算符时，你有两种选择：

选项 1：实现 OpenLineage 方法（推荐）

这是首选方法，可让你全面控制血缘提取：

from airflow.models import BaseOperator


class MyCustomOperator(BaseOperator):
    def __init__(self, source_table: str, target_table: str, **kwargs):
        super().__init__(**kwargs)
        self.source_table = source_table
        self.target_table = target_table

    def execute(self, context):
        # ... perform the actual work ...
        self.log.info(f"Processing {self.source_table} -> {self.target_table}")

    def get_openlineage_facets_on_complete(self, task_instance):
        """Return lineage after successful execution."""
        from openlineage.client.event_v2 import Dataset
        from airflow.providers.openlineage.extractors import OperatorLineage

        return OperatorLineage(
            inputs=[Dataset(namespace="warehouse://db", name=self.source_table)],
            outputs=[Dataset(namespace="warehouse://db", name=self.target_table)],
        )

选项 2：动态设置 Inlets/Outlets

对于较简单的场景，可在 execute 方法中设置血缘（仅适用于非延迟运算符）：

from airflow.models import BaseOperator
from openlineage.client.event_v2 import Dataset


class MyCustomOperator(BaseOperator):
    def __init__(self, source_table: str, target_table: str, **kwargs):
        super().__init__(**kwargs)
        self.source_table = source_table
        self.target_table = target_table

    def execute(self, context):
        # Set lineage dynamically based on operator parameters
        self.inlets = [
            Dataset(namespace="warehouse://db", name=self.source_table)
        ]
        self.outlets = [
            Dataset(namespace="warehouse://db", name=self.target_table)
        ]

        # ... perform the actual work ...
        self.log.info(f"Processing {self.source_table} -> {self.target_table}")

---

Dataset 命名辅助函数

使用 OpenLineage Dataset 命名辅助函数，确保不同平台上的命名一致：

from openlineage.client.event_v2 import Dataset

# Snowflake
from openlineage.client.naming.snowflake import SnowflakeDatasetNaming

naming = SnowflakeDatasetNaming(
    account_identifier="myorg-myaccount",
    database="mydb",
    schema="myschema",
    table="mytable",
)
dataset = Dataset(namespace=naming.get_namespace(), name=naming.get_name())
# -> namespace: "snowflake://myorg-myaccount", name: "mydb.myschema.mytable"

# BigQuery
from openlineage.client.naming.bigquery import BigQueryDatasetNaming

naming = BigQueryDatasetNaming(
    project="my-project",
    dataset="my_dataset",
    table="my_table",
)
dataset = Dataset(namespace=naming.get_namespace(), name=naming.get_name())
# -> namespace: "bigquery", name: "my-project.my_dataset.my_table"

# S3
from openlineage.client.naming.s3 import S3DatasetNaming

naming = S3DatasetNaming(bucket="my-bucket", key="path/to/file.parquet")
dataset = Dataset(namespace=naming.get_namespace(), name=naming.get_name())
# -> namespace: "s3://my-bucket", name: "path/to/file.parquet"

# PostgreSQL
from openlineage.client.naming.postgres import PostgresDatasetNaming

naming = PostgresDatasetNaming(
    host="localhost",
    port=5432,
    database="mydb",
    schema="public",
    table="users",
)
dataset = Dataset(namespace=naming.get_namespace(), name=naming.get_name())
# -> namespace: "postgres://localhost:5432", name: "mydb.public.users"

注意： 始终使用命名辅助函数，不要手动构造命名空间。如果你的平台缺少对应的辅助函数，请查看 OpenLineage 仓库或提交请求。

---

优先级规则

OpenLineage 按以下优先级提取血缘：

1. 自定义提取器（最高）- 用户注册的提取器
2. OpenLineage 方法 - 运算符中的 get_openlineage_facets_on_*
3. Hook 级血缘 - 通过 HookLineageCollector 从 hook 收集的血缘
4. Inlets/Outlets（最低）- 仅当其他方式都无法提取血缘时才会回退到此方式

注意： 如果提取器或方法存在但未返回数据集，OpenLineage 会检查 hook 级血缘，然后回退到 inlets/outlets。

---

最佳实践

使用命名辅助函数

始终使用 OpenLineage 命名辅助函数，以确保创建的数据集名称一致：

from openlineage.client.event_v2 import Dataset
from openlineage.client.naming.snowflake import SnowflakeDatasetNaming


def snowflake_dataset(schema: str, table: str) -> Dataset:
    """Create a Snowflake Dataset using the naming helper."""
    naming = SnowflakeDatasetNaming(
        account_identifier="mycompany",
        database="analytics",
        schema=schema,
        table=table,
    )
    return Dataset(namespace=naming.get_namespace(), name=naming.get_name())


# Usage
source = snowflake_dataset("raw", "orders")
target = snowflake_dataset("staging", "orders_clean")

记录血缘说明

添加注释，说明数据流：

transform = SqlOperator(
    task_id="transform_orders",
    sql="...",
    # Lineage: Reads raw orders, joins with customers, writes to staging
    inlets=[
        snowflake_dataset("raw", "orders"),
        snowflake_dataset("raw", "customers"),
    ],
    outlets=[
        snowflake_dataset("staging", "order_details"),
    ],
)

确保血缘准确

• SQL 查询变更时，更新 inlets/outlets
• 将 JOIN 中引用的所有表都列为 inlets
• 列出所有写入的表（包括相关的临时表）
• 仅标注 Outlet 或仅标注 Inlet 都是有效的。 即使其他 DAG 中没有对应的 inlet 或 outlet，也建议使用单向标注，以便查看血缘。

---

限制

限制	解决方法
仅支持表级（不支持列级血缘）	使用 OpenLineage 方法或自定义提取器
会被提取器或方法覆盖	仅用于没有提取器的运算符
在 DAG 解析时静态确定	在 execute() 中动态设置，或使用 OL 方法
可延迟运算符会丢失动态血缘	改用 OL 方法；在 execute() 中设置的属性会在延迟时丢失

---

相关技能

• creating-openlineage-extractors：用于列级血缘或复杂提取
• tracing-upstream-lineage：调查数据的来源
• tracing-downstream-lineage：调查依赖于数据的内容