dbt Core からメタデータをインポートする

このドキュメントでは、gcloud コマンドを使用して、dbt Core と MetricFlow から Knowledge Catalog(以前の Dataplex Universal Catalog)にメタデータをインポートする方法について説明します。

dbt インテグレーションによってキャプチャされるメタデータは次のとおりです。

  • 技術メタデータ: 主要なリソース(ソース、シード、モデル)とその技術プロパティ(列名、データ型、行数)が含まれます。
  • ビジネス メタデータとセマンティック メタデータ: dbt MetricFlow を活用し、セマンティック モデル、指標、保存済みクエリなどのビジネス定義とロジックが含まれます。
  • 運用メタデータとデータ品質メタデータ: タイミング、成功または失敗のステータス、データの更新速度、テストとテスト結果などの実行メタデータが含まれます。
  • リネージと関係のメタデータ: 変換グラフ(DAG)と dbt リソース間の依存関係、物理変換ブロックを追跡してリンクする物理リネージ、結合キーと動的結合、親子関係が含まれます。
  • 使用状況のメタデータ: dbt の外部でデータがどのように使用されているかをマッピングする、エクスポージャーでキャプチャされたメタデータが含まれます。

dbt Core と MetricFlow からメタデータをインポートする前に、次のタスクを完了します。

  1. 必要なロールと権限を付与します。
  2. Knowledge Catalog API を有効にします。
  3. dbt の前提条件を満たします。
  4. 宛先エントリ グループがまだ存在しない場合は、作成します。
  5. Cloud Storage のロールを理解する。

IAM のロールと権限

Knowledge Catalog コネクタ ジョブを作成して管理するには、Knowledge Catalog と Cloud Storage の権限を付与する Identity and Access Management(IAM)ロールが必要です。

dbt コネクタの構成に必要な権限を取得するには、管理者に次の IAM ロールを付与するよう依頼してください。

また、インポート ジョブがステージングされたメタデータ ファイルを読み取れるように、出力ステージング Cloud Storage バケット(--storage-uri)に対する Storage オブジェクト閲覧者(roles/storage.objectViewer)ロールを Knowledge Catalog サービス エージェント(service-PROJECT_NUMBER@gcp-sa-dataplex.iam.gserviceaccount.com)に付与する必要があります。

ロールの付与の詳細については、アクセスの管理をご覧ください。

API を有効にする

Knowledge Catalog API を有効にします。

API の有効化

dbt の前提条件

dbt メタデータの完全なセットをインポートするには、4 つの dbt JSON アーティファクト ファイルをすべて生成することをおすすめします。manifest.json のみが必要です。他のものはインポートを強化するもので、それらがなくても変換は正常に機能します。

  • manifest.json (必須): コア プロジェクトの構造と実行グラフ。MetricFlow のセマンティック モデル、指標、保存済みクエリも含まれます。
  • catalog.json: 列名とデータ型。catalog.json がないと、スキーマ アスペクトは型指定されていない列とともにインポートされます。
  • run_results.json: テスト結果と実行メタデータ。
  • sources.json: ソースの鮮度。

dbt メタデータ アーティファクト JSON ファイルの完全なセットを生成するには、次の dbt コマンドをこの順序で実行します。

  1. dbt source freshness
  2. dbt build
  3. dbt docs generate --no-compile

Cloud Storage のロールについて

dbt メタデータのインポートには、異なる目的で使用される 2 つの異なる Cloud Storage ロケーションが関係します。これらを混同しないでください。

  • 入力(dbt ソース アーティファクト): 生成された dbt JSON ファイルが存在する場所。これは、マシンまたは CI ランナー(./target/ や . など)のローカル ディレクトリ パス、または入力 Cloud Storage バケット URI 接頭辞(gs://my-dbt-artifacts-bucket/target/ など)にできます。このパスは --artifacts-path フラグを使用して指定します。gcloud コマンドは、ジョブの準備中にこれらの入力ファイルを読み取ります。Cloud Storage を使用する場合、gcloud コマンドを実行する呼び出し元には読み取りアクセス権(roles/storage.objectViewer または roles/storage.objectAdmin)が必要です。Knowledge Catalog サービス エージェントは、入力アーティファクト バケットにアクセスする必要はありません。
  • 出力(Knowledge Catalog インポート ステージング バケット): gcloud コマンドが変換されたメタデータ インポート ファイル(dbt_metadata.jsonl)をアップロードする Cloud Storage バケット URI 接頭辞(gs://my-staging-bucket/dbt-imports/ など)。Knowledge Catalog インポート ジョブは、取り込み中にこの接頭辞から読み取ります。この URI は --storage-uri フラグを使用して指定します。gcloud コマンドを実行する呼び出し元には、ファイルをアップロードするための書き込みアクセス権(roles/storage.objectCreator または roles/storage.objectAdmin)が必要です。また、Knowledge Catalog サービス エージェントには、ファイルをインポートするための読み取りアクセス権(roles/storage.objectViewer)が必要です。

dbt 接続を構成する

dbt 接続を確立するには、まず適切な dbt コマンドを実行してメタデータ アーティファクトを生成する必要があります。JSON ファイルが保存され、アクセス可能になると、インポート プロセスで次のアクションが実行されます。

  1. 入力アーティファクトを読み取る: dbt Core と MetricFlow によって生成された JSON アーティファクトを、入力の場所(--artifacts-path で指定されたローカル ディレクトリまたは Cloud Storage URI)から読み取ります。
  2. メタデータを変換する: コンテンツを Knowledge Catalog メタデータ インポート形式(dbt_metadata.jsonl)に変換します。
  3. ステージングにアップロード: 変換されたメタデータ インポート ファイルを、--storage-uri で指定された出力ステージングの Cloud Storage の場所にアップロードします。
  4. インポート ジョブをトリガーする: Knowledge Catalog サービス エージェントに、--storage-uri からステージングされたメタデータを読み取って Knowledge Catalog リソースに取り込むよう指示する Knowledge Catalog メタデータ インポート ジョブをトリガーします。

コンソール

  1. Google Cloud コンソールで、[Knowledge Catalog] > [コネクタ] ページに移動します。

    [コネクタ] に移動

  2. [Add connection] をクリックします。

  3. [コネクタ] リストで、[dbt Core と MetricFlow] カードを選択します。

  4. インポートした dbt アセットを表示するには、[検索] ページに移動するか、宛先のエントリ グループ ページを表示します。

gcloud

dbt メタデータ ジョブを作成する手順は次のとおりです。

  1. dbt メタデータ アーティファクト ファイルがローカルまたは入力 Cloud Storage バケットに保存されていることを確認します。
  2. 呼び出し元と Knowledge Catalog サービス エージェントの両方に適切な権限が構成された出力ステージング Cloud Storage バケットがあることを確認します。
  3. Cloud Shell、ローカル ターミナル、または自動ワークフロー ツールから、gcloud コマンドを実行します。

    gcloud alpha dataplex dbt metadata-jobs create my-dbt-import \
        --project=my-project \
        --location=us-central1 \
        --artifacts-path=. \
        --entry-group=dbt-metadata-ingestion \
        --storage-uri=gs://my-bucket/dbt-imports/
    

    必須フラグ

    • --storage-uri=STORAGE_URI: 変換された JSONL がアップロードされる (出力/ステージング)Cloud Storage URI 接頭辞(gs://bucket/path/)。インポート ジョブは、取り込み中にこの接頭辞から読み取ります。呼び出し元には書き込みアクセス権(roles/storage.objectCreator または roles/storage.objectAdmin)が、Knowledge Catalog サービス エージェントには読み取りアクセス権(roles/storage.objectViewer)が必要です。

    オプション フラグ

    • --artifacts-path=ARTIFACTS_PATH: (入力)移行元 dbt アーティファクトのパス。これは、ローカル ディレクトリ パス(. や ./target など)または Cloud Storage URI 接頭辞(gs://my-bucket/dbt-artifacts/ など)にできます。dbt プロジェクトのルート(target/ サブディレクトリは自動的に検出されます)または manifest.json を含むディレクトリを直接指定できます。デフォルトは . です。Cloud Storage URI が指定されている場合、呼び出し元は入力バケットに対する読み取りアクセス権(roles/storage.objectViewer または roles/storage.objectAdmin)を持っている必要があります。
    • --async: 処理中のオペレーションの完了を待たずに、直ちにコマンドを終了します。
    • --entry-group=ENTRY_GROUP: dbt エントリを受け取るエントリ グループの短い ID。プロジェクトとロケーションにすでに存在している必要があります(デフォルトは dbt-metadata-ingestion)。
    • --aspects-only: この dbt 実行で検出されたメタデータのみを更新し、エントリ グループの残りの部分は変更しません。エントリが作成、削除、再ペアレント化されることはありません。この実行で dbt アーティファクトがなかったアスペクトは、前の実行で与えられた値を保持します。これは、定期的な繰り返し取り込みに使用します。取り込みを再実行するをご覧ください。
    • --validate-only: JSON を作成してアップロードし、メタデータ ジョブを検証しますが、実際には取り込みません。
  4. ステータスが Created になっていることを確認します。

REST

REST API を使用して dbt メタデータをインポートするには:

  1. dbt アーティファクトを生成し、Knowledge Catalog JSON インポート ファイル(dbt_metadata.jsonl)に変換します。
  2. 変換されたファイルを Cloud Storage ステージング バケット(gs://BUCKET_NAME/PATH/)にアップロードします。
  3. projects.locations.metadataJobs.create メソッドを呼び出します。

    curl -X POST \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        -H "Content-Type: application/json" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/metadataJobs?metadataJobId=JOB_ID \
        -d '{
          "type": "IMPORT",
          "importSpec": {
            "sourceStorageUri": "gs://BUCKET_NAME/PATH/",
            "entrySyncMode": "FULL",
            "aspectSyncMode": "INCREMENTAL",
            "scope": {
              "entryGroups": [
                "projects/PROJECT_ID/locations/LOCATION/entryGroups/ENTRY_GROUP"
              ],
              "entryTypes": [
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-project",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-model",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-source",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-seed",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-snapshot",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-group",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-exposure",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-metric",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-macro",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-semantic-model",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-saved-query",
                "projects/dataplex-connector-types/locations/global/entryTypes/dbt-test"
              ],
              "aspectTypes": [
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-node",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-project",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-model",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-source",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-seed",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-snapshot",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-group",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-exposure",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-metric",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-macro",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-semantic-model",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-saved-query",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-data-quality",
                "projects/dataplex-connector-types/locations/global/aspectTypes/dbt-model-contracts"
              ]
            }
          }
        }'
    

    次のように置き換えます。

    • PROJECT_ID: エントリ グループが配置されている Google Cloud プロジェクト ID。
    • LOCATION: エントリ グループのリージョン(例: us-central1)。
    • JOB_ID: メタデータ ジョブの固有識別子。
    • BUCKET_NAME/PATH: dbt_metadata.jsonl がアップロードされた Cloud Storage URI 接頭辞。
    • ENTRY_GROUP: 宛先エントリ グループの短い ID。
  4. インポート ジョブのステータスを追跡するには、projects.locations.metadataJobs.get メソッドを使用します。

    curl -X GET \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/metadataJobs/JOB_ID
    

ジョブを作成すると、Knowledge Catalog は構成に従って初回実行をスケジュール設定するか、手動で開始できます。

取り込みを再実行する

最初のインポートの後、ほとんどの実行では、すでに存在するリソースのメタデータを更新するだけで済みます。これらの実行には --aspects-only を使用します。dbt 実行で検出されたもののみが更新され、エントリ グループ内の他のものはそのまま残されるため、任意のスケジュールで、複数のジョブから繰り返し実行しても安全です。

エントリのセットが変更された場合は、完全な取り込みを実行します(--aspects-only は省略)。

  • エントリ グループへの最初の取り込み。
  • dbt リソースが追加、名前変更、削除された。
  • エントリの表示名、説明、ラベルが変更された場合。
  • エントリ階層が変更されます。

フル実行では、ディスク上のアーティファクトからすべてのエントリの必要な側面が書き換えられるため、パイプラインで生成できる限り完全なアーティファクト セットから実行します。

定期的な更新には --aspects-only を実行します。

  • パイプラインが実行する dbt コマンド(dbt build、dbt test、dbt source freshness、または --select で絞り込んだ再ビルド)の後。
  • 列が追加、削除、再入力、再記述された。
  • モデルの SQL が変更され、実行時に catalog.json も書き込まれました。
  • 新しいテスト結果またはソースの鮮度。

--aspects-only はメタデータの追加と更新はできますが、削除はできません。

dbt メタデータを検索して表示する

コンソール

  1. Google Cloud コンソールで、[Knowledge Catalog] の [検索] ページに移動します。

    [検索] に移動

  2. [フィルタ] パネルで、dbt アセットをフィルタします。

    • [システム] セクションで、[インポートされたコンテキスト] を選択します。
    • 表示された [マネージド コネクタ] サブセクションで、[dbt] を選択します。
  3. 検索フィールドに、キーワード検索または自然言語検索を使用してクエリを入力します。たとえば、キーワード検索を使用してすべての dbt アセットを表示するには、「system=DBT」または「system=DBT AND type=dbt-model」と入力します。

  4. 検索結果で、任意の dbt アセットをクリックしてエントリの詳細ページを開き、スキーマ、リネージ、技術的な側面を表示します。

gcloud

  1. プロジェクト全体で dbt エントリを検索するには、gcloud dataplex entries search コマンドを使用します。

    gcloud dataplex entries search 'system=DBT' \
        --project=PROJECT_ID
    

    特定の dbt エントリ タイプ(モデルやソースなど)でフィルタするには:

    gcloud dataplex entries search 'system=DBT AND type=dbt-model' \
        --project=PROJECT_ID
    
  2. 特定の dbt エントリの詳細と側面を表示するには、gcloud dataplex entries lookup コマンドを使用します。

    gcloud dataplex entries lookup ENTRY_ID \
        --project=PROJECT_ID \
        --location=LOCATION \
        --entry-group=ENTRY_GROUP \
        --view=FULL
    

    次のように置き換えます。

    • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
    • LOCATION: エントリ グループのロケーション(例: us-central1)。
    • ENTRY_GROUP: 宛先エントリ グループの短い ID(例: dbt-metadata-ingestion)。
    • ENTRY_ID: dbt エントリの短い ID または相対リソース名。

REST

  1. dbt エントリを検索するには、projects.locations:searchEntries メソッドを呼び出します。

    curl -X POST \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        -H "Content-Type: application/json" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/global:searchEntries \
        -d '{
          "query": "system=DBT"
        }'
    

    特定の dbt リソースタイプでフィルタするには:

    curl -X POST \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        -H "Content-Type: application/json" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/global:searchEntries \
        -d '{
          "query": "system=DBT AND type=dbt-model"
        }'
    
  2. 特定のエントリの完全なメタデータ詳細とアスペクトを取得するには、projects.locations.entryGroups.entries.get メソッドを呼び出します。

    curl -X GET \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/entryGroups/ENTRY_GROUP/entries/ENTRY_ID?view=FULL
    
  3. 特定の dbt リソースの LLM コンテキストを取得するには、projects.locations:lookupContext API を使用します。

    curl -X POST \
        -H "Authorization: Bearer $(gcloud auth print-access-token)" \
        -H "Content-Type: application/json" \
        https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION:lookupContext \
        -d '{
          "resources": [
            "projects/PROJECT_ID/locations/LOCATION/entryGroups/ENTRY_GROUP/entries/ENTRY_ID"
          ]
        }'
    

    次のように置き換えます。

    • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
    • LOCATION: エントリ グループのロケーション(例: us-central1)。
    • ENTRY_GROUP: 宛先エントリ グループの短い ID(例: dbt-metadata-ingestion)。
    • ENTRY_ID: dbt エントリの短い ID または相対リソース名。

リソースの検索の詳細については、Knowledge Catalog でリソースを検索するをご覧ください。クエリ式とフィルタの詳細については、Knowledge Catalog の検索構文をご覧ください。

制限事項

  • 最近の dbt Core v1 バージョンをサポートします(バージョン 1.11 と 1.12 で検証済み)。dbt Core v2 と dbt Fusion は対象外です。
  • モデルのバージョニングを使用する dbt モデルは対象外です。
  • dbt Cloud はサポートされていません。
  • 非常に大きいスキーマやネストが深いスキーマは切り捨てられます。1 つのアスペクトがアスペクトあたりのサイズ上限を超えることはないため、ネストが深いスキーマでは末尾のフィールドが失われる可能性があります。
  • --aspects-only はメタデータの追加と更新はできますが、削除はできません。dbt リソースを削除するには、完全な実行が必要です。
  • エントリ リンクは対象外です。
  • この統合は、Data Lineage API とグラフの BigQuery リソースの dbt リネージ イベントのみをサポートします。外部のサードパーティ ソースの dbt エントリ(ソース、シード、モデル)は、データ リネージにキャプチャされません。

次のステップ