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

指導建立可重用的 Airflow DAG 範本並以 Pydantic 驗證，再用 YAML 組合 DAG。

Data & Analytics451今天更新

Weekly Brief

amplitude

產出 Amplitude 每週分析簡報，涵蓋趨勢、成果、風險與下週優先事項。

Data & Analytics42今天更新

User Cohort Forensics

amplitude

在 Amplitude 中調查個別使用者與使用者族群，解析身分、讀取個人資料並分析事件時間軸。

Data & Analytics42今天更新

Taxonomy

amplitude

Amplitude 事件分類體系參考指南：命名規範、資料品質稽核、評分與安全淘汰流程。

Data & Analytics42今天更新