自 2026 年 4 月 10 日起，Dataplex Universal Catalog 將更名為 Knowledge Catalog。API、用戶端程式庫、CLI 和 IAM 名稱維持不變。詳情請參閱「隆重推出 Google Cloud Knowledge Catalog」。

Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

查看 Google Cloud 系統的資料歷程

查看資料歷程，瞭解專案資源與建立這些資源的程序之間的關係。這些關係會顯示資料資產 (例如資料表和資料集) 如何透過查詢和管道等程序轉換。本指南說明如何在 Google Cloud 控制台中查看資料歷程詳細資料，或使用 Data Lineage API 擷取這些資料。

角色和權限

啟用 Data Lineage API 後，資料歷程會自動追蹤歷程資訊。您不需要任何管理員或編輯者角色，即可擷取資料資產的沿襲。

如要查看資料沿襲，您需要具備特定的 Identity and Access Management (IAM) 權限。系統會擷取多個專案的沿襲資訊，因此您需要多個專案的權限。

在 Knowledge Catalog、BigQuery 或 Vertex AI 中查看沿革時，您必須具備在查看沿革的專案中查看沿革資訊的權限。
查看其他專案中記錄的沿襲時：您需要有權限，才能查看這些專案中記錄的沿襲資訊。

如要取得查看資料歷程所需的權限，請要求管理員授予您下列 IAM 角色：

資料歷程檢視者 (roles/datalineage.viewer) 在記錄歷程的專案和查看歷程的專案中
查看 BigQuery 資料表詳細資料：資料表儲存專案的 BigQuery 資料檢視者 (roles/bigquery.dataViewer)
查看 BigQuery 工作詳細資料：工作運算專案的 BigQuery 資源檢視器 (roles/bigquery.resourceViewer)
查看其他編目資產的詳細資料：在儲存目錄項目的專案中，使用 Dataplex Catalog 檢視者 (roles/dataplex.catalogViewer)

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這些預先定義的角色具備查看資料沿襲所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

如要查看資料沿襲，必須具備下列權限：

查看 BigQuery 資料表詳細資料： bigquery.tables.get - 資料表的儲存空間專案
查看 BigQuery 工作詳細資料： bigquery.jobs.get - 工作的運算專案

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

資料歷程檢視畫面類型

您可以在 Google Cloud 控制台中，以互動式圖表或結構化清單的形式查看歷程資訊。

如要詳細瞭解圖表元素 (例如節點、邊緣、程序圖示和標籤) 和清單檢視畫面中可用的資料欄，請參閱「關於 Knowledge Catalog 中的資料沿襲視覺化」。

啟用資料歷程

啟用資料歷程，即可開始自動追蹤支援系統的歷程資訊。根據預設，啟用 API 後，系統會為大多數支援的服務啟動歷程追蹤功能。如要控管 Managed Service for Apache Spark 沿襲資料的擷取作業，請參閱「控管服務的沿襲資料擷取作業」。

您必須在查看歷程的專案和記錄歷程的專案中，啟用 Data Lineage API。詳情請參閱「專案類型」。

如要擷取沿襲資訊，請完成下列步驟：
1. 在 Google Cloud 控制台的「專案選取器」頁面中，選取要記錄歷程的專案。
  
  前往專案選取器
2. 啟用 Data Lineage API。
  
  啟用 API
3. 針對要記錄沿襲的每個專案，重複上述步驟。
在您查看歷程的專案中，啟用 Data Lineage API 和 Dataplex API。

啟用 API

控管服務的沿襲資料擷取作業

您可以在專案、資料夾或機構層級，選擇性啟用或停用特定服務的自動沿襲追蹤功能。

如要進一步瞭解如何透過資源樹狀結構以階層方式套用這些設定，請參閱「控管沿襲資料的擷取作業」。

必要條件

如要控管歷程擷取作業，請使用 Data Lineage API。請確認您已為用戶端專案設定計費和配額，因為 Data Lineage API 是以用戶端為基礎的 API。

在用戶端專案中啟用 datalineage.googleapis.com API。詳情請參閱「啟用資料沿襲」。
設定用戶端專案。在下列範例中，請使用 X-Goog-User-Project 標頭。詳情請參閱「系統參數」。

取得目前設定

如要檢查資源是否已啟用沿襲資料擷取功能，或在修改設定前取得 etag 值，請先擷取目前的設定。

C#

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 C# 設定說明操作。詳情請參閱 Data Lineage C# API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

using Google.Cloud.DataCatalog.Lineage.ConfigManagement.V1;

public sealed partial class GeneratedConfigManagementServiceClientSnippets
{
    /// <summary>Snippet for GetConfig</summary>
    /// <remarks>
    /// This snippet has been automatically generated and should be regarded as a code template only.
    /// It will require modifications to work:
    /// - It may require correct/in-range values for request initialization.
    /// - It may require specifying regional endpoints when creating the service client as shown in
    ///   https://cloud.google.com/dotnet/docs/reference/help/client-configuration#endpoint.
    /// </remarks>
    public void GetConfigRequestObject()
    {
        // Create client
        ConfigManagementServiceClient configManagementServiceClient = ConfigManagementServiceClient.Create();
        // Initialize request argument(s)
        GetConfigRequest request = new GetConfigRequest
        {
            ConfigName = ConfigName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
        };
        // Make the request
        Config response = configManagementServiceClient.GetConfig(request);
    }
}

Go

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Go 設定說明操作。詳情請參閱 Data Lineage Go API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。


//go:build examples

package main

import (
	"context"

	configmanagement "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1"
	configmanagementpb "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb"
)

func main() {
	ctx := context.Background()
	// This snippet has been automatically generated and should be regarded as a code template only.
	// It will require modifications to work:
	// - It may require correct/in-range values for request initialization.
	// - It may require specifying regional endpoints when creating the service client as shown in:
	//   https://pkg.go.dev/cloud.google.com/go#hdr-Client_Options
	c, err := configmanagement.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

	req := &configmanagementpb.GetConfigRequest{
		// TODO: Fill request struct fields.
		// See https://pkg.go.dev/cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb#GetConfigRequest.
	}
	resp, err := c.GetConfig(ctx, req)
	if err != nil {
		// TODO: Handle error.
	}
	// TODO: Use resp.
	_ = resp
}

Java

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Java 設定說明操作。詳情請參閱 Data Lineage Java API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;

public class SyncGetConfig {

  public static void main(String[] args) throws Exception {
    syncGetConfig();
  }

  public static void syncGetConfig() throws Exception {
    // This snippet has been automatically generated and should be regarded as a code template only.
    // It will require modifications to work:
    // - It may require correct/in-range values for request initialization.
    // - It may require specifying regional endpoints when creating the service client as shown in
    // https://cloud.google.com/java/docs/setup#configure_endpoints_for_the_client_library
    try (ConfigManagementServiceClient configManagementServiceClient =
        ConfigManagementServiceClient.create()) {
      GetConfigRequest request =
          GetConfigRequest.newBuilder()
              .setName(ConfigName.ofProjectLocationName("[PROJECT]", "[LOCATION]").toString())
              .build();
      Config response = configManagementServiceClient.getConfig(request);
    }
  }
}

Python

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Python 設定說明操作。詳情請參閱 Data Lineage Python API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

# This snippet has been automatically generated and should be regarded as a
# code template only.
# It will require modifications to work:
# - It may require correct/in-range values for request initialization.
# - It may require specifying regional endpoints when creating the service
#   client as shown in:
#   https://googleapis.dev/python/google-api-core/latest/client_options.html
from google.cloud import datacatalog_lineage_configmanagement_v1


def sample_get_config():
    # Create a client
    client = datacatalog_lineage_configmanagement_v1.ConfigManagementServiceClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_configmanagement_v1.GetConfigRequest(
        name="name_value",
    )

    # Make the request
    response = client.get_config(request=request)

    # Handle the response
    print(response)

gcloud

如要查看目前的沿襲設定，請使用 gcloud datalineage config describe 指令。您可以擷取專案、資料夾或機構的設定。

以下範例說明如何取得目前專案的設定：

gcloud datalineage config describe

舉例來說，如要取得特定專案的設定，請使用 --project 標記：

gcloud datalineage config describe --project=PROJECT_ID

更改下列內容：

PROJECT_ID：要查看設定的專案 ID。

如要查看資料夾或機構的服務目前沿襲擷取設定，請將 --project=PROJECT_ID 替換為：

--folder=FOLDER_ID，即可查看資料擷取設定。
--organization=ORGANIZATION_ID 如要查看機構的資料擷取設定。

REST

如要查看目前的沿襲設定，請使用 projects.locations.config.get 方法。您可以擷取專案、資料夾或機構的設定。

以下範例說明如何取得專案的設定：

使用任何要求資料之前，請先修改下列項目的值：

CLIENT_PROJECT_ID：用於帳單或配額的用戶端專案 ID。
PROJECT_ID：要查看設定的專案 ID。

HTTP 方法和網址：

GET https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

請展開以下其中一個選項，以傳送要求：

curl (Linux、macOS 或 Cloud Shell)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI，或已使用 Cloud Shell 自動登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

執行下列指令：

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

指令會傳回下列其中一個輸出內容：

如果未提供任何沿襲資料擷取設定，輸出內容會包含空白的 ingestion 物件：
```
{
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {}
}
  
```
也就是說，服務會使用預設的沿襲資料擷取設定。在本例中，Managed Service for Apache Spark 的沿襲資料擷取設定為 enabled。

如果您明確啟用沿襲資料擷取功能，就會看到下列輸出內容：

{
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

如果沿襲資料擷取功能已停用，您會看到下列輸出內容：

{
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

如要取得資料夾或機構的設定，請將 projects/PROJECT_ID 替換為 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

回應中的 etag 欄位是伺服器根據設定的目前值產生的檢查碼。使用 patch 方法更新設定時，您可以在要求主體中加入最近 get 要求傳回的 etag 值。如果您提供 etag，Knowledge Catalog 會使用該值驗證設定是否自上次讀取要求後有所變更。如果兩者不符，更新要求就會失敗。這樣一來，您就不會在讀取-修改-寫入情境中，意外覆寫其他使用者所做的設定。如果您未在 patch 要求中提供 etag，Knowledge Catalog 會無條件覆寫設定。

停用服務的沿襲資料擷取功能

如要管理費用、強制執行資料控管政策，或排除不適合使用沿襲追蹤功能的開發專案和其他工作負載，請停用服務的沿襲擷取功能。

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class DisableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    disableLineageIngestion(projectId, location);
  }

  // Disables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void disableLineageIngestion(String projectId, String location) throws Exception {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests.
    try (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to disable.
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to false to disable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(false).build();

      // Build the ingestion rule.
      IngestionRule disableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(disableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def disable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Disables lineage ingestion for a specific service.

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to disable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to False to disable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=False
    )

    # Create the ingestion rule.
    disable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )
     # Append the new disabling rule and assign it back to the config ingestion rules
    new_rules.append(disable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

如要為特定服務停用沿襲資料擷取功能，請使用 gcloud datalineage config update 指令，並搭配內嵌 JSON 字串或 JSON 檔案路徑，將特定 integration 的 lineageEnablement.enabled 設為 false。

以下範例說明如何使用內嵌 JSON 字串，為專案停用服務的沿襲資料擷取功能：

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": false
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

更改下列內容：

PROJECT_ID：要更新設定的專案 ID。
INTEGRATION：您要設定的整合服務。例如：DATAPROC 或 BIGQUERY。
ETAG：要求主體中最近一次 get 要求傳回的 etag 值，用於驗證自上次讀取要求以來，設定是否有所變更。

如要使用 JSON 檔案更新設定，請執行下列指令：

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

更改下列內容：

CONFIG_FILE：包含設定的 JSON 檔案路徑。

如要為資料夾或機構停用服務的沿襲資料擷取功能，請將 --project=PROJECT_ID 替換為：

--folder=FOLDER_ID 如要更新資料夾的資料擷取設定。
--organization=ORGANIZATION_ID 如果您想更新機構的資料擷取設定。

REST

如要針對特定服務停用沿襲資料擷取功能，請使用 projects.locations.config.patch 方法，並搭配將特定 integration 的 lineageEnablement.enabled 設為 false 的擷取規則。

為避免在讀取/修改/寫入情境中，意外覆寫其他使用者所做的設定，您可以在要求主體中加入 etag 欄位。詳情請參閱「取得目前設定」。

使用任何要求資料之前，請先修改下列項目的值：

CLIENT_PROJECT_ID：用於帳單或配額的用戶端專案 ID。
PROJECT_ID：要更新設定的專案 ID。
ETAG：最近 get 要求傳回的 etag 值。
INTEGRATION：您要設定的 integration。例如：DATAPROC。

HTTP 方法和網址：

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

JSON 要求主體：

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "ETAG"
}

請展開以下其中一個選項，以傳送要求：

curl (Linux、macOS 或 Cloud Shell)

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

您應該會收到如下的 JSON 回覆：

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

如要停用資料夾或機構的沿襲資料擷取功能，請將 projects/PROJECT_ID 替換為 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

為服務啟用歷程資料擷取功能

如要停用追蹤功能後繼續追蹤，或是啟用預設停用的整合功能，請為服務啟用沿襲資料擷取功能。

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class EnableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    enableLineageIngestion(projectId, location);
  }

  // Enables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void enableLineageIngestion(String projectId, String location) throws Exception {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests.
    try (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to enable (e.g., DATAPROC).
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to true to enable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(true).build();

      // Build the ingestion rule.
      IngestionRule enableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(enableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def enable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Enables lineage ingestion for a specific service like Dataproc
    (Managed Service for Apache Spark).

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to enable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to True to enable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=True
    )

    # Create the ingestion rule.
    enable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )

    # Append the new enabling rule and assign it back to the config ingestion rules
    new_rules.append(enable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

如要為特定服務啟用沿襲資料擷取功能，請使用 gcloud datalineage config update 指令，並搭配內嵌 JSON 字串或 JSON 檔案路徑，將特定 integration 的 lineageEnablement.enabled 設為 true。目前整合的服務包括 Managed Service for Apache Spark、BigQuery 和 Managed Airflow。

以下範例說明如何使用內嵌 JSON 字串，為專案啟用服務的沿襲資料擷取功能：

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

更改下列內容：

PROJECT_ID：要更新設定的專案 ID。
INTEGRATION：您要設定整合的項目 (例如 DATAPROC 或 BIGQUERY)。
ETAG：要求主體中最近一次 get 要求傳回的 etag 值，用於驗證自上次讀取要求以來，設定是否有所變更。

如要使用 JSON 檔案更新設定，請執行下列指令：

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

更改下列內容：

CONFIG_FILE：包含設定的 JSON 檔案路徑。

如要為資料夾或機構啟用服務的沿襲資料擷取功能，請將 --project=PROJECT_ID 替換為：

--folder=FOLDER_ID 如要更新資料夾的資料擷取設定。
--organization=ORGANIZATION_ID 如果您想更新機構的資料擷取設定。

REST

如要為特定服務啟用沿襲資料擷取功能，請使用 projects.locations.config.patch 方法和擷取規則，將特定 integration 的 lineageEnablement.enabled 設為 true。目前整合的服務包括 Managed Service for Apache Spark、BigQuery 和 Managed Airflow。

為避免在讀取/修改/寫入情境中，意外覆寫其他使用者所做的設定，您可以在要求主體中加入 etag 欄位。詳情請參閱「取得目前設定」。

使用任何要求資料之前，請先修改下列項目的值：

CLIENT_PROJECT_ID：用於帳單或配額的用戶端專案 ID。
PROJECT_ID：要更新設定的專案 ID。
ETAG：最近 get 要求傳回的 etag 值。
INTEGRATION：您要設定的 integration。例如：DATAPROC。

HTTP 方法和網址：

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

JSON 要求主體：

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "ETAG"
}

請展開以下其中一個選項，以傳送要求：

curl (Linux、macOS 或 Cloud Shell)

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "x-goog-user-project: CLIENT_PROJECT_ID" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config"

PowerShell (Windows)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred"; "x-goog-user-project" = "CLIENT_PROJECT_ID" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config" | Select-Object -Expand Content

您應該會收到如下的 JSON 回覆：

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

如要為資料夾或機構啟用服務的沿襲資料擷取功能，請將 projects/PROJECT_ID 替換為 folders/FOLDER_ID 或 organizations/ORGANIZATION_ID。

查看歷程

如要追蹤資料在系統間的轉換和移動方式，可以使用 Google Cloud 控制台或 API 查看資料歷程。

控制台

您可以在 Google Cloud 控制台中，從各種起點存取資料歷程資訊：

Knowledge Catalog：前往 Knowledge Catalog 的「Search」(搜尋) 頁面，選取「Knowledge Catalog」(知識目錄) 做為搜尋模式，搜尋要查看的項目，然後按一下該項目。詳情請參閱「在 Knowledge Catalog 中搜尋資源」。
BigQuery：前往 BigQuery 頁面，開啟要查看資料歷程的資料表。
Vertex AI：前往「資料集」或「模型登錄」頁面，然後點選要查看資料沿襲的資料集或模型。

如要查看沿襲圖，請按照下列步驟操作：

按一下「歷程」分頁標籤。

系統會開啟預設的「圖表」檢視畫面，顯示系統和區域的資料表層級歷程。詳情請參閱「沿襲圖表檢視畫面」。
如要手動探索沿襲圖，請按一下節點旁邊的「展開」，一次載入五個節點。

詳情請參閱「手動探索沿襲圖」。
在「圖表」檢視畫面中，點選節點。

系統會開啟「詳細資料」面板，顯示資產資訊，例如完整名稱和類型。詳情請參閱「節點詳細資料」。

注意： Vertex AI Pipelines 等系統會為 Vertex AI 模型和資料集生成沿襲資料。如要查看這些資產的詳細資料，請在「詳細資料」面板中點選「在 Vertex AI 中開啟」，直接在 Vertex AI 中查看沿襲圖詳細資料。
在「圖表」檢視畫面中，按一下含有程序圖示的邊緣。

「查詢」面板隨即開啟。詳情請參閱「檢查轉換邏輯」和「稽核及執行記錄」。
- 如要檢查轉換邏輯，請按一下「詳細資料」分頁標籤。
- 如要查看稽核記錄和執行作業記錄，請按一下「執行作業」分頁標籤。
注意： 在沿襲圖中，會顯示在 Vertex AI 處理程序 (例如元件和範本) 之前。如果程序已在 Knowledge Catalog 中編目，並具有完整網域名稱 (FQN)，您可以按一下該程序查看詳細資料，或按一下「詳細資料」面板中的「在 Vertex AI 中開啟」，在 Vertex AI 中查看圖表。如要查看更深入作業的歷程 (例如訓練資料、測試資料或套用的超參數)，請參閱「追蹤管道構件的歷程」。
在「沿襲關係探索器」面板中，選取篩選條件，例如「方向」、「依附元件類型」或「時間範圍」，然後按一下「套用」。

這會在特定區域中開啟焦點檢視畫面 (預覽)。這個檢視畫面會自動展開圖表，最多顯示三層節點。詳情請參閱「套用篩選器，取得焦點式沿革檢視畫面」。
在聚焦的「圖表」檢視畫面中選取節點，然後在節點的詳細資料面板中，按一下「Visualize Path」，即可從所選節點回溯至根項目，以視覺化方式呈現歷程路徑 (僅限聚焦檢視畫面)。

詳情請參閱「沿襲路徑視覺化」。
如要查看資料欄層級的沿襲 (僅適用於 BigQuery 和 Managed Service for Apache Spark 工作)，請執行下列其中一項操作：
- 在聚焦的「圖表」檢視畫面中，按一下表格中的「欄」圖示。
  
  「欄」圖示
- 在「歷程探索工具」面板中，依資料欄名稱篩選，然後按一下「套用」。
詳情請參閱資料欄層級歷程。
按一下「重設」。

這項操作會移除所有套用的篩選條件，並將您帶往圖表檢視畫面的開頭。
按一下「清單」即可切換至清單檢視畫面。

「清單」檢視畫面提供簡化且詳細的歷程表格表示法，適用於資料表和資料欄層級的歷程，並與「圖表」檢視畫面同步。根據預設，系統會顯示簡易清單檢視畫面，您可以切換至詳細清單檢視畫面，分析個別來源與目標之間的關係。您可以設定要顯示的資料欄，並匯出沿襲資料。詳情請參閱「沿革清單檢視畫面」。

Java

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.datacatalog.lineage.v1.BatchSearchLinkProcessesRequest;
import com.google.cloud.datacatalog.lineage.v1.EntityReference;
import com.google.cloud.datacatalog.lineage.v1.EventLink;
import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.LineageEvent;
import com.google.cloud.datacatalog.lineage.v1.Link;
import com.google.cloud.datacatalog.lineage.v1.ListLineageEventsRequest;
import com.google.cloud.datacatalog.lineage.v1.ListRunsRequest;
import com.google.cloud.datacatalog.lineage.v1.LocationName;
import com.google.cloud.datacatalog.lineage.v1.ProcessLinks;
import com.google.cloud.datacatalog.lineage.v1.Run;
import com.google.cloud.datacatalog.lineage.v1.SearchLinksRequest;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;

public class ViewLineageExample {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "my-project-id";
    String location = "us";
    String targetFullyQualifiedName = "bigquery:my-project-id.my_dataset.my_table";
    int maxDepth = 3;

    viewLineage(projectId, location, targetFullyQualifiedName, maxDepth);
  }

  static class Node {
    String fqn;
    int depth;
    Node(String fqn, int depth) {
      this.fqn = fqn;
      this.depth = depth;
    }
  }

  public static void viewLineage(
      String projectId, String location, String targetFullyQualifiedName, int maxDepth)
      throws IOException {
    // Initialize client that will be used to send requests. This client only needs
    // to be created once, and can be reused for multiple requests.
    try (LineageClient client = LineageClient.create()) {
      String parent = LocationName.of(projectId, location).toString();

      Set<String> visitedNodes = new HashSet<>();
      Queue<Node> queue = new LinkedList<>();

      visitedNodes.add(targetFullyQualifiedName);
      queue.offer(new Node(targetFullyQualifiedName, 0));

      while (!queue.isEmpty()) {
        Node current = queue.poll();
        System.out.printf("\nExploring node (Depth %d): %s\n", current.depth, current.fqn);

        if (current.depth >= maxDepth) {
          continue;
        }

        EntityReference targetEntity =
            EntityReference.newBuilder().setFullyQualifiedName(current.fqn).build();
        SearchLinksRequest searchLinksRequest =
            SearchLinksRequest.newBuilder().setParent(parent).setTarget(targetEntity).build();

        List<String> linkNames = new ArrayList<>();
        try {
          // 1. Search for links related to the target entity
          for (Link link : client.searchLinks(searchLinksRequest).iterateAll()) {
            linkNames.add(link.getName());
          }
        } catch (ApiException e) {
          System.out.printf("  Failed to retrieve links for %s: %s\n", current.fqn, e.getMessage());
          continue;
        }

        if (linkNames.isEmpty()) {
          continue;
        }

        // 2. Batch search for processes in chunks of 100
        for (int i = 0; i < linkNames.size(); i += 100) {
          List<String> batch = linkNames.subList(i, Math.min(linkNames.size(), i + 100));
          BatchSearchLinkProcessesRequest batchSearchRequest =
              BatchSearchLinkProcessesRequest.newBuilder()
                  .setParent(parent)
                  .addAllLinks(batch)
                  .build();

          try {
            for (ProcessLinks processLinks :
                client.batchSearchLinkProcesses(batchSearchRequest).iterateAll()) {
              String processName = processLinks.getProcess();
              System.out.printf("  Process: %s\n", processName);

              // 3. List runs for the process
              ListRunsRequest runsRequest =
                  ListRunsRequest.newBuilder().setParent(processName).build();
              for (Run run : client.listRuns(runsRequest).iterateAll()) {
                System.out.printf("    Run: %s\n", run.getName());

                // 4. List events for the run
                ListLineageEventsRequest eventsRequest =
                    ListLineageEventsRequest.newBuilder().setParent(run.getName()).build();
                for (LineageEvent event : client.listLineageEvents(eventsRequest).iterateAll()) {
                  for (EventLink eventLink : event.getLinksList()) {
                    String sourceFqn = eventLink.getSource().getFullyQualifiedName();
                    // If exploring upstream, queue the source
                    if (!sourceFqn.isEmpty() && !visitedNodes.contains(sourceFqn)) {
                      visitedNodes.add(sourceFqn);
                      queue.offer(new Node(sourceFqn, current.depth + 1));
                    }
                  }
                }
              }
            }
          } catch (ApiException e) {
            System.out.printf("  Failed to retrieve processes/runs: %s\n", e.getMessage());
          }
        }
      }
    }
  }
}

Python

from google.cloud import datacatalog_lineage_v1
from google.api_core.exceptions import GoogleAPICallError

def view_lineage(project_id: str, location: str, target_fully_qualified_name: str, max_depth: int = 3):
    """Retrieves lineage for a given entity using a depth-limited search."""
    client = datacatalog_lineage_v1.LineageClient()
    parent = f"projects/{project_id}/locations/{location}"

    # Store visited nodes to avoid infinite loops in cyclic graphs
    visited_nodes = set([target_fully_qualified_name])
    queue = [(target_fully_qualified_name, 0)]

    while queue:
        current_node, current_depth = queue.pop(0)
        print(f"\nExploring node (Depth {current_depth}): {current_node}")

        if current_depth >= max_depth:
            continue

        target_entity = datacatalog_lineage_v1.EntityReference(
            fully_qualified_name=current_node
        )
        search_links_request = datacatalog_lineage_v1.SearchLinksRequest(
            parent=parent,
            target=target_entity,
        )

        try:
            links = list(client.search_links(request=search_links_request))
        except GoogleAPICallError as e:
            print(f"  Failed to retrieve links for {current_node}: {e.message}")
            continue

        if not links:
            continue

        # Extract link names to query processes in batches
        link_names = [link.name for link in links]

        # Batch max size is 100
        for i in range(0, len(link_names), 100):
            batch = link_names[i:i + 100]
            batch_request = datacatalog_lineage_v1.BatchSearchLinkProcessesRequest(
                parent=parent,
                links=batch
            )

            try:
                for process_links in client.batch_search_link_processes(request=batch_request):
                    process_name = process_links.process
                    print(f"  Process: {process_name}")

                    runs_request = datacatalog_lineage_v1.ListRunsRequest(parent=process_name)
                    for run in client.list_runs(request=runs_request):
                        print(f"    Run: {run.name}")

                        events_request = datacatalog_lineage_v1.ListLineageEventsRequest(parent=run.name)
                        for event in client.list_lineage_events(request=events_request):
                            for event_link in event.links:
                                source_fqn = event_link.source.fully_qualified_name

                                # If exploring upstream, queue the source
                                if source_fqn and source_fqn not in visited_nodes:
                                    visited_nodes.add(source_fqn)
                                    queue.append((source_fqn, current_depth + 1))

            except GoogleAPICallError as e:
                 print(f"  Failed to retrieve processes/runs: {e.message}")

調整歷程圖表

如要調整沿襲關係圖，可以使用「沿襲關係探索器」中的醒目顯示和篩選選項：

如要搜尋特定專案、資料集或實體名稱，請使用「篩選器」面板。

套用篩選器後，符合篩選條件的沿襲節點會視為相符節點。您可以調整相符和不相符節點的顯示方式。
在沿襲圖中，按一下「清除篩選器」按鈕旁的「更多動作」圖示，即可查看顯示選項。
選取下列其中一個選項，或同時選取兩者：

你可以同時選取這兩個選項。如果同時選取這兩個選項，系統會隱藏未經過濾的節點，並在經過濾的圖表檢視中醒目顯示相符的節點。

後續步驟

追蹤 BigQuery 資料表副本和查詢工作的資料歷程。
瞭解資料歷程資訊模型。
瞭解資料歷程的注意事項和限制。
瞭解資料歷程稽核記錄。
瞭解如何排解資料歷程問題。
瞭解如何整合 OpenLineage。
瞭解如何搭配使用資料沿襲與 Managed Service for Apache Spark。

查看 Google Cloud 系統的資料歷程 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

角色和權限

所需權限

資料歷程檢視畫面類型

啟用資料歷程

控管服務的沿襲資料擷取作業

必要條件

取得目前設定

C#

C#

Go

Go

Java

Java

Python

Python

gcloud

REST

curl (Linux、macOS 或 Cloud Shell)

PowerShell (Windows)

停用服務的沿襲資料擷取功能

Java

Python

gcloud

REST

curl (Linux、macOS 或 Cloud Shell)

PowerShell (Windows)

為服務啟用歷程資料擷取功能

Java

Python

gcloud

REST

curl (Linux、macOS 或 Cloud Shell)

PowerShell (Windows)

查看歷程

控制台

Java

Python

調整歷程圖表

後續步驟

查看 Google Cloud 系統的資料歷程