tracing-upstream-lineage

上流リネージ：ソース

データの起源を追跡します。「このデータの出所はどこですか？」に答えます。

リネージの調査

ステップ 1：対象の種類を特定する

何を追跡するかを決めます。

• テーブル：このテーブルにデータを取り込む処理を追跡します
• 列：この特定の列の出所を追跡します
• DAG：この DAG が読み取るデータソースを追跡します

ステップ 2：データを生成する DAG を見つける

通常、テーブルには Airflow DAG によってデータが取り込まれます。関連付けを確認します。

1. 名前で DAG を検索する：af dags list を使い、テーブル名に一致する DAG 名を探します

   • load_customers -> customers テーブル
   • etl_daily_orders -> orders テーブル

2. DAG のソースコードを調べる：af dags source <dag_id> を使って DAG の定義を読み取ります

   • INSERT、MERGE、CREATE TABLE 文を探します
   • コード内で対象テーブルを見つけます

3. DAG のタスクを確認する：af tasks list <dag_id> を使って、DAG が実行する操作を確認します

Astro の場合

Astro を実行している場合は、Astro UI の Lineage タブで、DAG やデータセットをまたいだリネージを視覚的に調べられます。DAG のソースコードを手動で検索せずに、上流の依存関係をすばやく追跡できます。

OSS Airflow の場合

DAG のソースコードとタスクログを使ってリネージを追跡します（DAG 間を横断する UI は組み込まれていません）。

ステップ 3：データソースを追跡する

DAG のコードから、ソーステーブルとシステムを特定します。

SQL ソース（FROM 句を探します）：

# In DAG code:
SELECT * FROM source_schema.source_table  # <- This is an upstream source

外部ソース（接続の参照を探します）：

• S3Operator -> S3 バケットのソース
• PostgresOperator -> Postgres データベースのソース
• SalesforceOperator -> Salesforce API のソース
• HttpOperator -> REST API のソース

ファイルソース：

• オブジェクトストレージ内の CSV／Parquet ファイル
• SFTP 経由で配置されたファイル
• ローカルのファイルパス

ステップ 4：リネージの連鎖を構築する

各ソースを再帰的に追跡します。

TARGET: analytics.orders_daily
    ^
    +-- DAG: etl_daily_orders
            ^
            +-- SOURCE: raw.orders (table)
            |       ^
            |       +-- DAG: ingest_orders
            |               ^
            |               +-- SOURCE: Salesforce API (external)
            |
            +-- SOURCE: dim.customers (table)
                    ^
                    +-- DAG: load_customers
                            ^
                            +-- SOURCE: PostgreSQL (external DB)

ステップ 5：ソースの健全性を確認する

各上流ソースについて確認します。

• テーブル：checking-freshness スキルで鮮度を確認します
• DAG：af dags stats で最近の実行状況を確認します
• 外部システム：DAG のコードから接続情報を記録します

列のリネージ

特定の列を追跡する場合：

1. 対象テーブルのスキーマで列を見つけます
2. その列名が参照されている箇所を DAG のソースコードで検索します
3. 変換処理をたどります。
   • 直接マッピング：source.col AS target_col
   • 変換：COALESCE(a.col, b.col) AS target_col
   • 集計：SUM(detail.amount) AS total_amount

出力：リネージレポート

概要

一文で回答します。「このテーブルには、ソース Y と Z から DAG X によってデータが取り込まれます」

リネージ図

[Salesforce] --> [raw.opportunities] --> [stg.opportunities] --> [fct.sales]
                        |                        |
                   DAG: ingest_sfdc         DAG: transform_sales

ソースの詳細

ソース	種類	接続	鮮度	所有者
raw.orders	テーブル	内部	2時間前	data-team
Salesforce	API	salesforce_conn	リアルタイム	sales-ops

変換の連鎖

データの流れと変換について説明します。

1. Salesforce API の同期によって、生データが raw.orders に取り込まれます
2. DAG transform_orders がデータをクレンジングして重複を除去し、stg.orders に格納します
3. DAG build_order_facts がディメンションと結合し、fct.orders に格納します

データ品質への影響

• 単一障害点はありますか？
• 上流ソースが古くなっていませんか？
• 破損する可能性のある複雑な変換の連鎖はありますか？

関連スキル

• ソースの鮮度を確認する：checking-freshness スキル
• ソース DAG をデバッグする：debugging-dags スキル
• 下流への影響を追跡する：tracing-downstream-lineage スキル
• リネージの注釈を手動で追加する：annotating-task-lineage スキル
• カスタムのリネージ抽出ツールを作成する：creating-openlineage-extractors スキル