上流の系統: ソース

データの起源を追跡する - 「このデータはどこから来ているのか?」に答える

系統調査

ステップ 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句を探す):

# DAGコード内:
SELECT * FROM source_schema.source_table  # <- これが上流ソース

外部ソース (接続参照を探す):

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

ファイルソース:

• オブジェクトストレージ内のCSV/Parquetファイル
• SFTPドロップ
• ローカルファイルパス

ステップ 4: 系統チェーンを構築

各ソースを再帰的に追跡:

ターゲット: analytics.orders_daily
    ^
    +-- DAG: etl_daily_orders
            ^
            +-- ソース: raw.orders (テーブル)
            |       ^
            |       +-- DAG: ingest_orders
            |               ^
            |               +-- ソース: Salesforce API (外部)
            |
            +-- ソース: dim.customers (テーブル)
                    ^
                    +-- DAG: load_customers
                            ^
                            +-- ソース: PostgreSQL (外部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

出力: 系統レポート

サマリー

1行の答え: 「このテーブルはDAG Xによってソース Y と Z から入力されている」

系統図

[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 スキル