Annotating Task Lineage

作者 astronomercbe1141f547b无许可证451 个星标收录于 2026年10月8日更新于 2026年10月8日仓库今天更新

Annotate Airflow tasks with data lineage using inlets and outlets. Use when the user wants to add lineage metadata to tasks, specify input/output datasets, or enable lineage tracking for operators without built-in OpenLineage extraction.

仅含说明Data & Analytics
AI 生成的概览

指导使用 inlets 和 outlets 为 Airflow 任务添加手动数据血缘标注。

功能
说明如何通过 inlets 和 outlets 为 Airflow 任务标注输入与输出数据集，涵盖 OpenLineage Dataset 对象、Airflow Asset 和 Airflow Dataset。内容包括基础算子用法、多输入多输出、在自定义算子中动态设置血缘，以及 Snowflake、BigQuery、S3 和 PostgreSQL 的数据集命名辅助工具。还记录了优先级规则、最佳实践和局限，例如仅支持表级血缘，以及可延迟算子会丢失动态血缘。
适用场景
适用于为 Airflow 任务添加血缘元数据、指定输入或输出数据集，或为没有内置 OpenLineage 提取能力的算子启用血缘跟踪。它面向只需表级血缘、希望无需自定义提取代码即可快速配置的场景。若需要列级血缘或复杂提取逻辑，则应改用 OpenLineage 方法或自定义提取器。
运行要求
不附带脚本，仅为说明文档。所述用法假定已有 Airflow 环境并安装 OpenLineage provider（openlineage.client.event_v2、airflow.providers.openlineage），部分示例还需要 Airflow 3 的 Asset 或 Airflow 2.4 及以上的 Dataset。阅读该技能本身不需要凭据或网络访问。

Annotating Task Lineage with Inlets & Outlets

This skill guides you through adding manual lineage annotations to Airflow tasks using inlets and outlets.

Reference: See the OpenLineage provider developer guide for the latest supported operators and patterns.

On Astro

Lineage annotations defined with inlets and outlets are visualized in Astro's enhanced Lineage tab, which provides cross-DAG and cross-deployment lineage views. This means your annotations are immediately visible in the Astro UI, giving you a unified view of data flow across your entire Astro organization.

When to Use This Approach

ScenarioUse Inlets/Outlets?
Operator has OpenLineage methods (get_openlineage_facets_on_*)❌ Modify the OL method directly
Operator has no built-in OpenLineage extractor✅ Yes
Simple table-level lineage is sufficient✅ Yes
Quick lineage setup without custom code✅ Yes
Need column-level lineage❌ Use OpenLineage methods or custom extractor
Complex extraction logic needed❌ Use OpenLineage methods or custom extractor

Note: Inlets/outlets are the lowest-priority fallback. If an OpenLineage extractor or method exists for the operator, it takes precedence. Use this approach for operators without extractors.

Supported Types for Inlets/Outlets

You can use OpenLineage Dataset objects or Airflow Assets for inlets and outlets:

OpenLineage Datasets (Recommended)

python
from openlineage.client.event_v2 import Dataset
# Database tablessource_table = Dataset(    namespace="postgres://mydb:5432",    name="public.orders",)target_table = Dataset(    namespace="snowflake://account.snowflakecomputing.com",    name="staging.orders_clean",)
# Filesinput_file = Dataset(    namespace="s3://my-bucket",    name="raw/events/2024-01-01.json",)

Airflow Assets (Airflow 3+)

python
from airflow.sdk import Asset
# Using Airflow's native Asset typeorders_asset = Asset(uri="s3://my-bucket/data/orders")

Airflow Datasets (Airflow 2.4+)

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

Basic Usage

Setting Inlets and Outlets on Operators

python
from airflow import DAGfrom airflow.operators.bash import BashOperatorfrom openlineage.client.event_v2 import Datasetimport pendulum
# Define your lineage datasetssource_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

Multiple Inputs and Outputs

Tasks often read from multiple sources and write to multiple destinations:

python
from openlineage.client.event_v2 import Dataset
# Multiple source tablescustomers = 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 tablesdaily_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)

Setting Lineage in Custom Operators

When building custom operators, you have two options:

Option 1: Implement OpenLineage Methods (Recommended)

This is the preferred approach as it gives you full control over lineage extraction:

python
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)],        )

Option 2: Set Inlets/Outlets Dynamically

For simpler cases, set lineage within the execute method (non-deferrable operators only):

python
from airflow.models import BaseOperatorfrom 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 Naming Helpers

Use the OpenLineage dataset naming helpers to ensure consistent naming across platforms:

python
from openlineage.client.event_v2 import Dataset
# Snowflakefrom 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"
# BigQueryfrom 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"
# S3from 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"
# PostgreSQLfrom 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"

Note: Always use the naming helpers instead of constructing namespaces manually. If a helper is missing for your platform, check the OpenLineage repo or request it.

Precedence Rules

OpenLineage uses this precedence for lineage extraction:

  1. Custom Extractors (highest) - User-registered extractors
  2. OpenLineage Methods - get_openlineage_facets_on_* in operator
  3. Hook-Level Lineage - Lineage collected from hooks via HookLineageCollector
  4. Inlets/Outlets (lowest) - Falls back to these if nothing else extracts lineage

Note: If an extractor or method exists but returns no datasets, OpenLineage will check hook-level lineage, then fall back to inlets/outlets.

Best Practices

Use the Naming Helpers

Always use OpenLineage naming helpers for consistent dataset creation:

python
from openlineage.client.event_v2 import Datasetfrom 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())
# Usagesource = snowflake_dataset("raw", "orders")target = snowflake_dataset("staging", "orders_clean")

Document Your Lineage

Add comments explaining the data flow:

python
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"),    ],)

Keep Lineage Accurate

  • Update inlets/outlets when SQL queries change
  • Include all tables referenced in JOINs as inlets
  • Include all tables written to (including temp tables if relevant)
  • Outlet-only and inlet-only annotations are valid. One-sided annotations are encouraged for lineage visibility even without a corresponding inlet or outlet in another DAG.

Limitations

LimitationWorkaround
Table-level only (no column lineage)Use OpenLineage methods or custom extractor
Overridden by extractors/methodsOnly use for operators without extractors
Static at DAG parse timeSet dynamically in execute() or use OL methods
Deferrable operators lose dynamic lineageUse OL methods instead; attributes set in execute() are lost when deferring

Related Skills

  • creating-openlineage-extractors: For column-level lineage or complex extraction
  • tracing-upstream-lineage: Investigate where data comes from
  • tracing-downstream-lineage: Investigate what depends on data

来源与署名

来源：astronomer/agents位于skills/annotating-task-lineage提交cbe1141

许可证： 无许可证

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架

更多来自 astronomer/agents 的技能

Warehouse Init

astronomer

发现数据仓库的模式元数据，并生成可共享的 .astro/warehouse.md 参考文件，以便快速进行概念到表的查找。

Data & Analytics451今天更新

Tracing Upstream Lineage

astronomer

Trace upstream data lineage. Use when the user asks where data comes from, what feeds a table, upstream dependencies, data sources, or needs to understand data origins.

待分类451今天更新

Tracing Downstream Lineage

astronomer

追踪下游数据血缘，评估表或 DAG 变更会造成哪些影响。

Data & Analytics451今天更新

Testing Dags

astronomer

指导使用 af 命令行工具对 Airflow DAG 进行迭代式测试、调试与修复。

Software Development451今天更新

Setting Up Astro Project

astronomer

使用 Astro CLI 初始化并配置 Astro/Airflow 项目，涵盖依赖、连接与变量。

DevOps & Cloud451今天更新

Profiling Tables

astronomer

通过 SQL 统计、基数分析、样本数据和数据质量评估，对数据库表进行剖析。

Data & Analytics451今天更新

更多Data & Analytics技能

Creating Openlineage Extractors

astronomer

指导为缺少内置血缘支持的 Airflow 算子创建自定义 OpenLineage 提取器。

Data & Analytics451今天更新

Checking Freshness

astronomer

通过查询时间戳列检查数据库表的数据新鲜度，并报告过期状态。

Data & Analytics451今天更新

Blueprint

astronomer

指导构建带 Pydantic 校验的可复用 Airflow DAG 模板，并用 YAML 组合 DAG。

Data & Analytics451今天更新

Weekly Brief

amplitude

生成 Amplitude 每周分析简报，涵盖趋势、成果、风险及下周优先事项。

Data & Analytics42今天更新

User Cohort Forensics

amplitude

在 Amplitude 中调查单个用户与用户群体，解析身份、读取资料并分析事件时间线。

Data & Analytics42今天更新

Taxonomy

amplitude

Amplitude 事件分类体系参考指南：命名规范、数据质量审计、评分与安全弃用流程。

Data & Analytics42今天更新