annotating-task-lineage

Inlets と Outlets を使ったタスクリネージの注釈付け

このスキルでは、inlets と outlets を使って Airflow タスクに手動でリネージ注釈を追加する方法を説明します。

この方法を使う場面

シナリオ	Inlets/Outlets を使いますか？
オペレーターに OpenLineage メソッドがある	いいえ。OL メソッドを直接変更します
オペレーターに組み込みの OpenLineage エクストラクターがない	はい
テーブルレベルのシンプルなリネージで十分	はい
カスタムコードを使わず、すばやくリネージを設定したい	はい
カラムレベルのリネージが必要	いいえ。OpenLineage メソッドまたはカスタムエクストラクターを使います
複雑な抽出ロジックが必要	いいえ。OpenLineage メソッドまたはカスタムエクストラクターを使います

Inlets/Outlets でサポートされている型

OpenLineage データセット（推奨）

from openlineage.client.event_v2 import Dataset

source_table = Dataset(
    namespace="postgres://mydb:5432",
    name="public.orders",
)

Airflow アセット（Airflow 3+）

from airflow.sdk import Asset

orders_asset = Asset(uri="s3://my-bucket/data/orders")

Airflow データセット（Airflow 2.4+）

from airflow.datasets import Dataset

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

source_table = Dataset(namespace="snowflake://account", name="raw.orders")
target_table = Dataset(namespace="snowflake://account", 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],
        outlets=[target_table],
    )

    export = BashOperator(
        task_id="export_to_s3",
        bash_command="echo 'exporting...'",
        inlets=[target_table],
        outlets=[output_file],
    )

    transform >> export

複数の入力と出力

from openlineage.client.event_v2 import Dataset

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")

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],
    outlets=[daily_summary, customer_metrics],
)

カスタムオペレーター

オプション 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):
        self.log.info(f"Processing {self.source_table} -> {self.target_table}")

    def get_openlineage_facets_on_complete(self, task_instance):
        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 を動的に設定する

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):
        self.inlets = [Dataset(namespace="warehouse://db", name=self.source_table)]
        self.outlets = [Dataset(namespace="warehouse://db", name=self.target_table)]