從 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 連接器工作,您需要 Identity and Access Management (IAM) 角色,授予 Knowledge Catalog 和 Cloud Storage 的權限。

如要取得設定 dbt 連接器所需的權限,請要求管理員授予下列 IAM 角色:

  • 如要建立及管理項目群組,您必須在專案中具備 Dataplex Catalog 管理員 (roles/dataplex.catalogAdmin)、Dataplex Catalog 編輯者 (roles/dataplex.catalogEditor) 或 Dataplex 項目群組擁有者 (roles/dataplex.entryGroupOwner) 角色。
  • 如要執行 dbt gcloud 指令並建立中繼資料匯入作業: 請遵循最低權限原則,授予下列角色:

    或者,您也可以在專案中授予「Dataplex Catalog 管理員」(roles/dataplex.catalogAdmin) 角色和「Dataplex 中繼資料工作擁有者」(roles/dataplex.metadataJobOwner) 角色。

  • 如要將轉換後的中繼資料上傳至輸出暫存 bucket (--storage-uri): 暫存 bucket 的「Storage 物件建立者」(roles/storage.objectCreator) 或「Storage 物件管理員」(roles/storage.objectAdmin)。

  • 如要從輸入 Cloud Storage bucket 讀取 dbt 構件 (如果使用 Cloud Storage,則為 --artifacts-path): 輸入構件 bucket 的「Storage 物件檢視者」(roles/storage.objectViewer) 或「Storage 物件管理員」(roles/storage.objectAdmin)。如果您具備 Storage 物件管理員角色,則不需要 Storage 物件檢視者角色。

  • 如要查看 dbt 中繼資料: 在專案中,指派 Dataplex Catalog 檢視者 (roles/dataplex.catalogViewer) 角色。

  • 如要在 Cloud Logging 中查看記錄,請在專案中啟用「記錄檢視者」(roles/logging.viewer)。

此外,您必須在輸出暫存 Cloud Storage bucket (--storage-uri) 中,將「Storage 物件檢視者」(roles/storage.objectViewer) 角色授予 Knowledge Catalog 服務代理 (service-PROJECT_NUMBER@gcp-sa-dataplex.iam.gserviceaccount.com),匯入作業才能讀取暫存的中繼資料檔案。

如要進一步瞭解如何授予角色,請參閱「管理存取權」。

啟用 API

啟用 Knowledge Catalog API。

啟用 API

dbt 必要條件

如要匯入整組 dbt 中繼資料,建議您產生所有四個 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

    前述 dbt build 的測試結果會遺失,且目錄會將所有資料品質測試回報為通過。

瞭解 Cloud Storage 角色

匯入 dbt 中繼資料時,會用到兩個不同的 Cloud Storage 位置,用途各異,不應混淆:

  • 輸入 (dbt 來源構件):產生 dbt JSON 檔案的位置。可以是機器或 CI 執行器上的本機目錄路徑 (例如 ./target/ 或 .),也可以是輸入 Cloud Storage bucket URI 前置字串 (例如 gs://my-dbt-artifacts-bucket/target/)。您可以使用 --artifacts-path 旗標提供這個路徑。gcloud 指令會在準備工作時讀取這些輸入檔案。如果使用 Cloud Storage,執行 gcloud 指令的呼叫端需要讀取存取權 (roles/storage.objectViewer 或 roles/storage.objectAdmin)。Knowledge Catalog 服務代理不需要存取輸入構件 bucket。
  • 輸出內容 (Knowledge Catalog 匯入暫存 bucket):Cloud Storage bucket URI 前置字元 (例如 gs://my-staging-bucket/dbt-imports/),gcloud 指令會將轉換後的中繼資料匯入檔案 (dbt_metadata.jsonl) 上傳至該位置,Knowledge Catalog 匯入工作也會在擷取期間從該位置讀取資料。您可以使用 --storage-uri 旗標提供這個 URI。執行 gcloud 指令的呼叫端需要寫入權限 (roles/storage.objectCreator 或 roles/storage.objectAdmin) 才能上傳檔案,而 Knowledge Catalog 服務代理程式需要讀取權限 (roles/storage.objectViewer) 才能匯入檔案。

設定 dbt 連線

如要建立 dbt 連線,請先執行適當的 dbt 指令,生成中繼資料構件。JSON 檔案儲存完畢並可供存取後,匯入程序會執行下列動作:

  1. 讀取輸入構件:從輸入位置 (--artifacts-path 中指定的本機目錄或 Cloud Storage URI),讀取 dbt Core 和 MetricFlow 產生的 JSON 構件。
  2. 轉換中繼資料:將內容轉換為 Knowledge Catalog 中繼資料匯入格式 (dbt_metadata.jsonl)。
  3. 上傳至暫存位置:將轉換後的中繼資料匯入檔案上傳至 --storage-uri 中指定的輸出暫存 Cloud Storage 位置。
  4. 觸發匯入作業:觸發 Knowledge Catalog 中繼資料匯入作業,指示 Knowledge Catalog 服務代理程式從 --storage-uri 讀取並擷取暫存中繼資料,然後匯入 Knowledge Catalog 資源。

控制台

  1. 前往 Google Cloud 控制台的「Knowledge Catalog」「Connectors」頁面。

    前往「連線器」

  2. 按一下「新增連線」。

  3. 在「連結器」清單中,選取「dbt Core 和 MetricFlow」資訊卡。

  4. 如要查看匯入的 dbt 資產,請前往「搜尋」頁面,或查看目的地「項目群組」頁面。

gcloud

如要建立 dbt 中繼資料工作,請完成下列步驟:

  1. 確認 dbt 中繼資料構件檔案儲存在本機或輸入 Cloud Storage 值區中。
  2. 請確認您已設定輸出暫存 Cloud Storage bucket,並為呼叫端和 Knowledge Catalog 服務代理程式授予適當權限。
  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:(輸出/暫存) Cloud Storage URI 前置字元 (gs://bucket/path/),轉換後的 JSONL 會上傳至此處,匯入作業也會從這裡讀取資料。呼叫端必須具備寫入存取權 (roles/storage.objectCreator 或 roles/storage.objectAdmin),知識目錄服務代理必須具備讀取存取權 (roles/storage.objectViewer)。

    選用旗標

    • --artifacts-path=ARTIFACTS_PATH:(輸入) 來源 dbt 構件的路徑。可以是本機目錄路徑 (例如 . 或 ./target),也可以是 Cloud Storage URI 前置字串 (例如 gs://my-bucket/dbt-artifacts/)。可以指向 dbt 專案根目錄 (系統會自動偵測 target/ 子目錄),也可以直接指向包含 manifest.json 的目錄。預設值為 .。如果提供 Cloud Storage URI,呼叫端必須具備輸入 bucket 的讀取權限 (roles/storage.objectViewer 或 roles/storage.objectAdmin)。
    • --async:立即返回,不要等待執行中的作業完成。
    • --entry-group=ENTRY_GROUP:接收 dbt 項目項目的項目群組簡短 ID。必須已存在於專案和位置 (預設為 dbt-metadata-ingestion)。
    • --aspects-only:只更新這個 dbt 執行作業觀察到的中繼資料,並保留其餘項目群組不變。系統不會建立、刪除或重新設定父項目的項目,且 dbt 構件缺少的層面會保留先前執行作業的值。適用於例行性、重複攝取。請參閱「重新執行擷取作業」。
    • --validate-only:建構及上傳 JSON,並驗證中繼資料作業,但實際上不會擷取資料。
  4. 確認您收到「已建立」狀態。

REST

如要使用 REST API 匯入 dbt 中繼資料,請按照下列步驟操作:

  1. 產生 dbt 構件,並轉換為 Knowledge Catalog JSON 匯入檔案 (dbt_metadata.jsonl)。
  2. 將轉換後的檔案上傳至 Cloud Storage 暫存 bucket (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:中繼資料工作的專屬 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」「Search」(搜尋) 頁面。

    前往「搜尋」

  2. 在「Filters」(篩選器) 面板中,篩選 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。
  • 系統會截斷過大或深度巢狀結構的結構定義:單一切面不得超過每個切面的大小上限,因此深度巢狀結構的結構定義可能會遺失尾端欄位。
  • --aspects-only可以新增及重新整理中繼資料,但無法移除。 刪除 dbt 資源需要完整執行。
  • 不支援進入連結。
  • 這項整合功能僅支援 Data Lineage API 和圖表中的 BigQuery 資源 dbt 沿襲事件。外部第三方來源的 dbt 項目 (來源、種子、模型) 不會擷取至資料沿襲。

後續步驟