本文說明如何使用 gcloud 指令,將 dbt Core 和 MetricFlow 的中繼資料匯入 Knowledge Catalog (舊稱 Dataplex Universal Catalog)。
dbt 整合服務會擷取下列中繼資料:
- 技術中繼資料:包括重要資源 (來源、種子、模型) 及其技術屬性 (資料欄名稱、資料類型、列數)。
- 業務和語意中繼資料:由 dbt MetricFlow 提供支援,包括語意模型、指標和已儲存查詢等業務定義和邏輯。
- 作業和資料品質中繼資料:包括執行中繼資料,例如時間、成功或失敗狀態、資料更新間隔、測試和測試結果。
- 沿襲和關係中繼資料:包括轉換圖 (DAG) 和 dbt 資源之間的依附元件、追蹤及連結實體轉換區塊的實體沿襲、聯結鍵和動態聯結,以及父項與子項關係。
- 取用中繼資料:包括曝光次數中擷取的中繼資料,可對應 dbt 以外的資料使用方式。
如要從 dbt Core 和 MetricFlow 匯入中繼資料,請先完成下列工作:
- 授予必要的角色和權限。
- 啟用 Knowledge Catalog API。
- 符合 dbt 必要條件。
- 建立目的地項目群組 (如果還沒有的話)。
- 瞭解 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 中繼資料工作擁有者 (
roles/dataplex.metadataJobOwner)。 - 目標項目群組或專案的 Dataplex 項目群組匯入者 (
roles/dataplex.entryGroupImporter)。
或者,您也可以在專案中授予「Dataplex Catalog 管理員」(
roles/dataplex.catalogAdmin) 角色和「Dataplex 中繼資料工作擁有者」(roles/dataplex.metadataJobOwner) 角色。- 專案的 Dataplex 中繼資料工作擁有者 (
如要將轉換後的中繼資料上傳至輸出暫存 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。
dbt 必要條件
如要匯入整組 dbt 中繼資料,建議您產生所有四個 dbt JSON 構件檔案。只需要 manifest.json,其他屬性則可豐富匯入內容,即使沒有這些屬性,轉換作業也會正常運作:
manifest.json(必要):核心專案結構和執行圖表。也包含 MetricFlow 語意模型、指標和已儲存的查詢。catalog.json:資料欄名稱和資料類型。如果沒有catalog.json,系統會匯入結構定義層面,但資料欄沒有型別。run_results.json: 測試結果和執行中繼資料。sources.json: 來源新鮮度。
如要產生完整的 dbt 中繼資料構件 JSON 檔案集,可以依序執行下列 dbt 指令:
dbt source freshnessdbt build
前述dbt docs generate --no-compiledbt 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 檔案儲存完畢並可供存取後,匯入程序會執行下列動作:
- 讀取輸入構件:從輸入位置 (
--artifacts-path中指定的本機目錄或 Cloud Storage URI),讀取 dbt Core 和 MetricFlow 產生的 JSON 構件。 - 轉換中繼資料:將內容轉換為 Knowledge Catalog 中繼資料匯入格式 (
dbt_metadata.jsonl)。 - 上傳至暫存位置:將轉換後的中繼資料匯入檔案上傳至
--storage-uri中指定的輸出暫存 Cloud Storage 位置。 - 觸發匯入作業:觸發 Knowledge Catalog 中繼資料匯入作業,指示 Knowledge Catalog 服務代理程式從
--storage-uri讀取並擷取暫存中繼資料,然後匯入 Knowledge Catalog 資源。
控制台
前往 Google Cloud 控制台的「Knowledge Catalog」「Connectors」頁面。
按一下「新增連線」。
在「連結器」清單中,選取「dbt Core 和 MetricFlow」資訊卡。
如要查看匯入的 dbt 資產,請前往「搜尋」頁面,或查看目的地「項目群組」頁面。
gcloud
如要建立 dbt 中繼資料工作,請完成下列步驟:
- 確認 dbt 中繼資料構件檔案儲存在本機或輸入 Cloud Storage 值區中。
- 請確認您已設定輸出暫存 Cloud Storage bucket,並為呼叫端和 Knowledge Catalog 服務代理程式授予適當權限。
從 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,並驗證中繼資料作業,但實際上不會擷取資料。
確認您收到「已建立」狀態。
REST
如要使用 REST API 匯入 dbt 中繼資料,請按照下列步驟操作:
- 產生 dbt 構件,並轉換為 Knowledge Catalog JSON 匯入檔案 (
dbt_metadata.jsonl)。 - 將轉換後的檔案上傳至 Cloud Storage 暫存 bucket (
gs://BUCKET_NAME/PATH/)。 呼叫
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。
如要追蹤匯入工作的狀態,請使用
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 中繼資料
控制台
前往 Google Cloud 控制台的「Knowledge Catalog」「Search」(搜尋) 頁面。
在「Filters」(篩選器) 面板中,篩選 dbt 資產:
- 在「系統」部分,選取「匯入的內容」。
- 在隨即顯示的「受管理連接器」子區段中,選取「dbt」。
在搜尋欄位中,使用關鍵字或自然語言搜尋輸入查詢。舉例來說,如要使用關鍵字搜尋查看所有 dbt 資產,請輸入
system=DBT或system=DBT AND type=dbt-model。在搜尋結果中,按一下任一 dbt 資產,開啟該項目的詳細資料頁面,即可查看其結構、歷程和技術層面。
gcloud
如要在專案中搜尋 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如要查看特定 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
如要搜尋 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" }'如要擷取特定項目的完整中繼資料詳細資料和層面,請呼叫
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如要擷取特定 dbt 資源的 LLM 內容,請使用
projects.locations:lookupContextAPI: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 項目 (來源、種子、模型) 不會擷取至資料沿襲。
- 如要在 Data Lineage API 中擷取所有 dbt 歷程事件,請使用 OpenLineage dbt 整合功能。 然後將 OpenLineage 與 Knowledge Catalog 整合,從 dbt 匯入及視覺化呈現資料歷程。
後續步驟
- 瞭解如何管理連結器工作。