Creating Openlineage Extractors

作者 astronomercbe1141f547b無授權條款451 個星標收錄於 2026年10月8日更新於 2026年10月8日儲存庫今天更新

Create custom OpenLineage extractors for Airflow operators. Use when the user needs lineage from unsupported or third-party operators, wants column-level lineage, or needs complex extraction logic beyond what inlets/outlets provide.

僅含說明Software DevelopmentData & Analytics
AI 產生的概覽

指導為缺少內建血緣支援的 Airflow 算子建立自訂 OpenLineage 擷取器。

功能
此技能說明如何從沒有內建 OpenLineage 支援的 Airflow 算子擷取資料血緣。內容涵蓋兩種做法：在自有算子上直接加入 OpenLineage 方法，以及為無法修改的第三方算子撰寫 BaseExtractor 子類別。它提供 SQL 算子、檔案傳輸與執行時決定血緣的程式碼範例，並說明註冊、測試、優先順序規則與常見陷阱。
適用情境
當需要從不受支援或第三方 Airflow 算子取得血緣、需要欄層級血緣，或擷取邏輯超出 inlets 與 outlets 所能提供時使用。也適用於在 OpenLineage 方法與自訂擷取器之間做選擇。
執行需求
需要安裝 OpenLineage provider 的 Apache Airflow 以及 openlineage 用戶端程式庫；擷取器必須能從 Airflow worker 匯入，並透過 airflow.cfg 或環境變數註冊。此技能不附帶指令碼，僅包含說明與程式碼範例。

Creating OpenLineage Extractors

This skill guides you through creating custom OpenLineage extractors to capture lineage from Airflow operators that don't have built-in support.

Reference: See the OpenLineage provider developer guide for the latest patterns and list of supported operators/hooks.

When to Use Each Approach

ScenarioApproach
Operator you own/maintainOpenLineage Methods (recommended, simplest)
Third-party operator you can't modifyCustom Extractor
Need column-level lineageOpenLineage Methods or Custom Extractor
Complex extraction logicOpenLineage Methods or Custom Extractor
Simple table-level lineageInlets/Outlets (simplest, but lowest priority)

Important: Always prefer OpenLineage methods over custom extractors when possible. Extractors are harder to write, easier to diverge from operator behavior after changes, and harder to debug.

On Astro

Astro includes built-in OpenLineage integration — no additional transport configuration is needed. Lineage events are automatically collected and displayed in the Astro UI's Lineage tab. Custom extractors deployed to an Astro project are automatically picked up, so you only need to register them in airflow.cfg or via environment variable and deploy.

Two Approaches

1. OpenLineage Methods (Recommended)

Use when you can add methods directly to your custom operator. This is the go-to solution for operators you own.

2. Custom Extractors

Use when you need lineage from third-party or provider operators that you cannot modify.

Approach 1: OpenLineage Methods (Recommended)

When you own the operator, add OpenLineage methods directly:

python
from airflow.models import BaseOperator
class MyCustomOperator(BaseOperator):    """Custom operator with built-in OpenLineage support."""
    def __init__(self, source_table: str, target_table: str, **kwargs):        super().__init__(**kwargs)        self.source_table = source_table        self.target_table = target_table        self._rows_processed = 0  # Set during execution
    def execute(self, context):        # Do the actual work        self._rows_processed = self._process_data()        return self._rows_processed
    def get_openlineage_facets_on_start(self):        """Called when task starts. Return known inputs/outputs."""        # Import locally to avoid circular imports        from openlineage.client.event_v2 import Dataset        from airflow.providers.openlineage.extractors import OperatorLineage
        return OperatorLineage(            inputs=[Dataset(namespace="postgres://db", name=self.source_table)],            outputs=[Dataset(namespace="postgres://db", name=self.target_table)],        )
    def get_openlineage_facets_on_complete(self, task_instance):        """Called after success. Add runtime metadata."""        from openlineage.client.event_v2 import Dataset        from openlineage.client.facet_v2 import output_statistics_output_dataset        from airflow.providers.openlineage.extractors import OperatorLineage
        return OperatorLineage(            inputs=[Dataset(namespace="postgres://db", name=self.source_table)],            outputs=[                Dataset(                    namespace="postgres://db",                    name=self.target_table,                    facets={                        "outputStatistics": output_statistics_output_dataset.OutputStatisticsOutputDatasetFacet(                            rowCount=self._rows_processed                        )                    },                )            ],        )
    def get_openlineage_facets_on_failure(self, task_instance):        """Called after failure. Optional - for partial lineage."""        return None

OpenLineage Methods Reference

MethodWhen CalledRequired
get_openlineage_facets_on_start()Task enters RUNNINGNo
get_openlineage_facets_on_complete(ti)Task succeedsNo
get_openlineage_facets_on_failure(ti)Task failsNo

Implement only the methods you need. Unimplemented methods fall through to Hook-Level Lineage or inlets/outlets.

Approach 2: Custom Extractors

Use this approach only when you cannot modify the operator (e.g., third-party or provider operators).

Basic Structure

python
from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineagefrom openlineage.client.event_v2 import Dataset
class MyOperatorExtractor(BaseExtractor):    """Extract lineage from MyCustomOperator."""
    @classmethod    def get_operator_classnames(cls) -> list[str]:        """Return operator class names this extractor handles."""        return ["MyCustomOperator"]
    def _execute_extraction(self) -> OperatorLineage | None:        """Called BEFORE operator executes. Use for known inputs/outputs."""        # Access operator properties via self.operator        source_table = self.operator.source_table        target_table = self.operator.target_table
        return OperatorLineage(            inputs=[                Dataset(                    namespace="postgres://mydb:5432",                    name=f"public.{source_table}",                )            ],            outputs=[                Dataset(                    namespace="postgres://mydb:5432",                    name=f"public.{target_table}",                )            ],        )
    def extract_on_complete(self, task_instance) -> OperatorLineage | None:        """Called AFTER operator executes. Use for runtime-determined lineage."""        # Access properties set during execution        # Useful for operators that determine outputs at runtime        return None

OperatorLineage Structure

python
from airflow.providers.openlineage.extractors.base import OperatorLineagefrom openlineage.client.event_v2 import Datasetfrom openlineage.client.facet_v2 import sql_job
lineage = OperatorLineage(    inputs=[Dataset(namespace="...", name="...")],      # Input datasets    outputs=[Dataset(namespace="...", name="...")],     # Output datasets    run_facets={"sql": sql_job.SQLJobFacet(query="SELECT...")},  # Run metadata    job_facets={},                                      # Job metadata)

Extraction Methods

MethodWhen CalledUse For
_execute_extraction()Before operator runsStatic/known lineage
extract_on_complete(task_instance)After successRuntime-determined lineage
extract_on_failure(task_instance)After failurePartial lineage on errors

Registering Extractors

Option 1: Configuration file (airflow.cfg)

ini
[openlineage]extractors = mypackage.extractors.MyOperatorExtractor;mypackage.extractors.AnotherExtractor

Option 2: Environment variable

bash
AIRFLOW__OPENLINEAGE__EXTRACTORS='mypackage.extractors.MyOperatorExtractor;mypackage.extractors.AnotherExtractor'

Important: The path must be importable from the Airflow worker. Place extractors in your DAGs folder or installed package.

Common Patterns

SQL Operator Extractor

python
from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineagefrom openlineage.client.event_v2 import Datasetfrom openlineage.client.facet_v2 import sql_job
class MySqlOperatorExtractor(BaseExtractor):    @classmethod    def get_operator_classnames(cls) -> list[str]:        return ["MySqlOperator"]
    def _execute_extraction(self) -> OperatorLineage | None:        sql = self.operator.sql        conn_id = self.operator.conn_id
        # Parse SQL to find tables (simplified example)        # In practice, use a SQL parser like sqlglot        inputs, outputs = self._parse_sql(sql)
        namespace = f"postgres://{conn_id}"
        return OperatorLineage(            inputs=[Dataset(namespace=namespace, name=t) for t in inputs],            outputs=[Dataset(namespace=namespace, name=t) for t in outputs],            job_facets={                "sql": sql_job.SQLJobFacet(query=sql)            },        )
    def _parse_sql(self, sql: str) -> tuple[list[str], list[str]]:        """Parse SQL to extract table names. Use sqlglot for real parsing."""        # Simplified example - use proper SQL parser in production        inputs = []        outputs = []        # ... parsing logic ...        return inputs, outputs

File Transfer Extractor

python
from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineagefrom openlineage.client.event_v2 import Dataset
class S3ToSnowflakeExtractor(BaseExtractor):    @classmethod    def get_operator_classnames(cls) -> list[str]:        return ["S3ToSnowflakeOperator"]
    def _execute_extraction(self) -> OperatorLineage | None:        s3_bucket = self.operator.s3_bucket        s3_key = self.operator.s3_key        table = self.operator.table        schema = self.operator.schema
        return OperatorLineage(            inputs=[                Dataset(                    namespace=f"s3://{s3_bucket}",                    name=s3_key,                )            ],            outputs=[                Dataset(                    namespace="snowflake://myaccount.snowflakecomputing.com",                    name=f"{schema}.{table}",                )            ],        )

Dynamic Lineage from Execution

python
from openlineage.client.event_v2 import Dataset
class DynamicOutputExtractor(BaseExtractor):    @classmethod    def get_operator_classnames(cls) -> list[str]:        return ["DynamicOutputOperator"]
    def _execute_extraction(self) -> OperatorLineage | None:        # Only inputs known before execution        return OperatorLineage(            inputs=[Dataset(namespace="...", name=self.operator.source)],        )
    def extract_on_complete(self, task_instance) -> OperatorLineage | None:        # Outputs determined during execution        # Access via operator properties set in execute()        outputs = self.operator.created_tables  # Set during execute()
        return OperatorLineage(            inputs=[Dataset(namespace="...", name=self.operator.source)],            outputs=[Dataset(namespace="...", name=t) for t in outputs],        )

Common Pitfalls

1. Circular Imports

Problem: Importing Airflow modules at the top level causes circular imports.

python
# ❌ BAD - can cause circular import issuesfrom airflow.models import TaskInstancefrom openlineage.client.event_v2 import Dataset
class MyExtractor(BaseExtractor):    ...
python
# ✅ GOOD - import inside methodsclass MyExtractor(BaseExtractor):    def _execute_extraction(self):        from openlineage.client.event_v2 import Dataset        # ...

2. Wrong Import Path

Problem: Extractor path doesn't match actual module location.

bash
# ❌ Wrong - path doesn't existAIRFLOW__OPENLINEAGE__EXTRACTORS='extractors.MyExtractor'
# ✅ Correct - full importable pathAIRFLOW__OPENLINEAGE__EXTRACTORS='dags.extractors.my_extractor.MyExtractor'

3. Not Handling None

Problem: Extraction fails when operator properties are None.

python
# ✅ Handle optional propertiesdef _execute_extraction(self) -> OperatorLineage | None:    if not self.operator.source_table:        return None  # Skip extraction
    return OperatorLineage(...)

Testing Extractors

Unit Testing

python
import pytestfrom unittest.mock import MagicMockfrom mypackage.extractors import MyOperatorExtractor
def test_extractor():    # Mock the operator    operator = MagicMock()    operator.source_table = "input_table"    operator.target_table = "output_table"
    # Create extractor    extractor = MyOperatorExtractor(operator)
    # Test extraction    lineage = extractor._execute_extraction()
    assert len(lineage.inputs) == 1    assert lineage.inputs[0].name == "input_table"    assert len(lineage.outputs) == 1    assert lineage.outputs[0].name == "output_table"

Precedence Rules

OpenLineage checks for lineage in this order:

  1. Custom Extractors (highest priority)
  2. OpenLineage Methods on operator
  3. Hook-Level Lineage (from HookLineageCollector)
  4. Inlets/Outlets (lowest priority)

If a custom extractor exists, it overrides built-in extraction and inlets/outlets.

Related Skills

  • annotating-task-lineage: For simple table-level lineage with inlets/outlets
  • tracing-upstream-lineage: Investigate data origins
  • tracing-downstream-lineage: Investigate data dependencies

來源與署名

來源：astronomer/agents位於skills/creating-openlineage-extractors提交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今天更新

更多Software Development技能

Writing Plans

obra

Use when you have a spec or requirements for a multi-step task, before touching code

Software Development296K今天更新

Verification Before Completion

obra

要求在聲稱工作已完成、已修正或已通過之前，先執行最新的驗證指令並確認輸出。

Software Development296K今天更新

Using Git Worktrees

obra

為功能開發建立隔離的 git worktree 工作區，並執行專案初始化與基線測試。

Software Development296K今天更新

Requesting Code Review

obra

派發程式碼審查子代理，並為它準備精確的上下文，在問題擴散前先找出問題。

Software Development296K今天更新

Receiving Code Review

obra

指導代理接收程式碼審查意見：核實建議、釐清不明項目、以技術理由反駁，並逐項實作修正。

Software Development296K今天更新

Finishing A Development Branch

obra

引導代理收尾已完成的開發分支：驗證測試、選擇合併、開 PR 或保留，然後清理工作區。

Software Development296K今天更新