mcpField
C
creating-openlineage-extractors

creating-openlineage-extractors

github.com/necatiarslan数据与数据库4446 次安装21 周前更新
代码仓库

关于 creating-openlineage-extractors

用于构建自定义 OpenLineage 提取器的技能，可捕获 Airflow 中尚无内置支持的算子血缘信息。内容涵盖 OpenLineage 方法、自定义提取器及简易的 inlets 和 outlets，并介绍方案选择与注册方式。

主要功能

  1. 01根据场景选择提取方案
  2. 02在算子中实现 OpenLineage 方法
  3. 03构建自定义算子提取器
  4. 04通过配置或环境变量注册提取器

适用场景

  1. 01捕获不受支持的 Airflow 算子血缘
  2. 02添加列级血缘提取
  3. 03使用 inlets 和 outlets 记录简单表级血缘

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/necatiarslan/airflow-vscode-extension --skill creating-openlineage-extractors

安装后在 Claude Code 中输入/creating-openlineage-extractors确认已加载。

详细介绍

创建 OpenLineage 提取器

本技能将指导你创建自定义 OpenLineage 提取器，以捕获尚未内置支持的 Airflow 运算符中的血缘信息。

何时使用各方法

场景方法
你拥有或维护的运算符OpenLineage 方法（推荐）
无法修改的第三方运算符自定义提取器
需要列级血缘信息OpenLineage 方法或自定义提取器
复杂的提取逻辑OpenLineage 方法或自定义提取器
简单的表级血缘信息Inlets/Outlets（最简单，优先级最低）

方法 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
        self._rows_processed = 0

    def execute(self, context):
        self._rows_processed = self._process_data()
        return self._rows_processed

    def get_openlineage_facets_on_start(self):
        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):
        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
                        )
                    },
                )
            ],
        )

方法 2：自定义提取器

from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineage
from openlineage.client.event_v2 import Dataset

class MyOperatorExtractor(BaseExtractor):
    @classmethod
    def get_operator_classnames(cls) -> list[str]:
        return ["MyCustomOperator"]

    def _execute_extraction(self) -> OperatorLineage | None:
        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:
        return None

注册提取器

配置文件：

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

环境变量：

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

评论

支持 Markdown