使用伺服器端自動化功能搜尋多區域沿襲

searchLineageStreaming API 可讓您查詢多層級、跨區域的資料歷程，並以即時串流的形式接收統合圖表。您可以建構工具，周遊廣泛、深入或多專案資料架構，而不會發生要求逾時。詳情請參閱「關於多區域歷程搜尋」。

主要功能

searchLineageStreaming API 包含下列功能：

  • 廣度優先搜尋：從一組定義的根實體，逐層向上或向下遍歷歷程圖，準確計算每個連結資產的深度。

  • 串流回覆：後端系統發現子圖和歷程連結時，會即時傳送包含這些內容的分塊回覆。標準歷程查詢 API 在大型多專案圖表上可能會逾時，但串流功能可避免在廣泛或深層圖表上發生要求逾時。

  • 多位置和多專案遍歷：雖然您在要求路徑中只指定一個計費專案，但只要具備必要權限，API 就會自動探索及遍歷多個 Google Cloud 專案和地理位置的歷程連結。

  • 精細的資料欄層級歷程：支援搜尋資產之間的資料欄層級依附元件。

  • 萬用字元查詢：在完整名稱 (FQN) 後方加上 *，即可擷取特定實體的所有資料欄層級歷程。

  • 管道深入分析：可選擇擷取轉換管道 (程序) 的中繼資料，這些管道建立了歷程連結。

事前準備

向 API 發出要求前，請確認您已符合下列安全性和環境必要條件：

必要的角色

如要取得搜尋資料歷程連結所需的權限，請要求管理員在儲存歷程連結和程序的專案中，授予您資料歷程檢視者 (roles/datalineage.viewer) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備搜尋資料歷程連結所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

必須具備下列權限，才能搜尋資料歷程連結：

  • 搜尋實體層級歷程： datalineage.events.get 儲存連結的專案
  • 搜尋資料欄層級歷程： datalineage.events.getFields 在儲存連結的專案中
  • 擷取完整管道程序詳細資料：在儲存程序的專案上執行 datalineage.processes.get

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

資源範圍

設定 API 要求時，您必須區分用於管理計費的資源，以及 API 掃描的實際位置：

  • 帳單上層路徑：網址要求中的 parent 路徑必須採用 projects/project/locations/location 格式。這組特定的專案和位置配對，僅用於評估計費配額和 API 頻率限制。

  • 指定地區：在要求主體的 locations 陣列中，明確定義要讓後端掃描的區域。

驗證設定

使用 Google Cloud 存取權杖初始化環境變數，驗證 curl 指令：

export ACCESS_TOKEN=$(gcloud auth print-access-token)

使用範例

以下範例使用端點 datalineage.googleapis.com。

搜尋多層級、多專案歷程

如要執行深度歷程搜尋，遍歷圖表的各個深度，並掃描不同的 Google Cloud 專案，請定義下列變數：

  • 將 limits.maxDepth 設為目標遍歷深度 (接受 1 到 100 之間的值)。

  • 在 locations 陣列中填入要讓後端交叉比對的目標區域 (例如 ["us", "us-east1"])。

C#

C#

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

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

using Google.Api.Gax.Grpc;
using Google.Api.Gax.ResourceNames;
using Google.Cloud.DataCatalog.Lineage.V1;
using System.Threading.Tasks;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for SearchLineageStreaming</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 async Task SearchLineageStreamingRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        SearchLineageStreamingRequest request = new SearchLineageStreamingRequest
        {
            ParentAsLocationName = LocationName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
            Locations = { "", },
            RootCriteria = new SearchLineageStreamingRequest.Types.RootCriteria(),
            Direction = SearchLineageStreamingRequest.Types.SearchDirection.Unspecified,
            Filters = new SearchLineageStreamingRequest.Types.SearchFilters(),
            Limits = new SearchLineageStreamingRequest.Types.SearchLimits(),
        };
        // Make the request, returning a streaming response
        using LineageClient.SearchLineageStreamingStream response = lineageClient.SearchLineageStreaming(request);

        // Read streaming responses from server until complete
        // Note that C# 8 code can use await foreach
        AsyncResponseStream<SearchLineageStreamingResponse> responseStream = response.GetResponseStream();
        while (await responseStream.MoveNextAsync())
        {
            SearchLineageStreamingResponse responseItem = responseStream.Current;
            // Do something with streamed response
        }
        // The response stream has completed
    }
}

Java

Java

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

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

import com.google.api.gax.rpc.ServerStream;
import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.LocationName;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingRequest;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingResponse;
import java.util.ArrayList;

public class AsyncSearchLineageStreaming {

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

  public static void asyncSearchLineageStreaming() 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 (LineageClient lineageClient = LineageClient.create()) {
      SearchLineageStreamingRequest request =
          SearchLineageStreamingRequest.newBuilder()
              .setParent(LocationName.of("[PROJECT]", "[LOCATION]").toString())
              .addAllLocations(new ArrayList<String>())
              .setRootCriteria(SearchLineageStreamingRequest.RootCriteria.newBuilder().build())
              .setFilters(SearchLineageStreamingRequest.SearchFilters.newBuilder().build())
              .setLimits(SearchLineageStreamingRequest.SearchLimits.newBuilder().build())
              .build();
      ServerStream<SearchLineageStreamingResponse> stream =
          lineageClient.searchLineageStreamingCallable().call(request);
      for (SearchLineageStreamingResponse response : stream) {
        // Do something when a response is received.
      }
    }
  }
}

Node.js

Node.js

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Node.js 設定操作說明進行操作。詳情請參閱 Data Lineage Node.js 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.
 * TODO(developer): Uncomment these variables before running the sample.
 */
/**
 *  Required. The project and location to initiate the search from.
 */
// const parent = 'abc123'
/**
 *  Required. The locations to search in.
 */
// const locations = ['abc','def']
/**
 *  Required. Criteria for the root of the search.
 */
// const rootCriteria = {}
/**
 *  Required. Direction of the search.
 */
// const direction = {}
/**
 *  Optional. Filters for the search.
 */
// const filters = {}
/**
 *  Optional. Limits for the search.
 */
// const limits = {}

// Imports the Lineage library
const {LineageClient} = require('@google-cloud/lineage').v1;

// Instantiates a client
const lineageClient = new LineageClient();

async function callSearchLineageStreaming() {
  // Construct request
  const request = {
    parent,
    locations,
    rootCriteria,
    direction,
  };

  // Run request
  const stream = await lineageClient.searchLineageStreaming(request);
  stream.on('data', (response) => { console.log(response) });
  stream.on('error', (err) => { throw(err) });
  stream.on('end', () => { /* API call completed */ });
}

callSearchLineageStreaming();

Python

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_v1


def sample_search_lineage_streaming():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.SearchLineageStreamingRequest(
        parent="parent_value",
        locations=["locations_value1", "locations_value2"],
        direction="UPSTREAM",
    )

    # Make the request
    stream = client.search_lineage_streaming(request=request)

    # Handle the response
    for response in stream:
        print(response)

小茹

Ruby

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

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

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the search_lineage_streaming call in the Lineage service
#
# 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/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#search_lineage_streaming.
#
def search_lineage_streaming
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingRequest.new

  # Call the search_lineage_streaming method to start streaming.
  output = client.search_lineage_streaming request

  # The returned object is a streamed enumerable yielding elements of type
  # ::Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingResponse
  output.each do |current_response|
    p current_response
  end
end

REST

如要搜尋資料歷程，請使用 searchLineageStreaming 方法。

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

  • PROJECT_ID：用於管理帳單和配額評估的 Google Cloud 專案 ID。
  • LOCATION_ID： Google Cloud 位置，例如 us-central1。
  • SOURCE_PROJECT_ID：來源資料表所在的 Google Cloud 專案 ID。
  • DATASET_ID：BigQuery 資料集 ID。
  • TABLE_ID：BigQuery 資料表 ID。

HTTP 方法和網址：

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

JSON 要求主體：

{
  "parent": "projects/PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID",
    "us-east1",
    "us-central1"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "bigquery:SOURCE_PROJECT_ID.DATASET_ID.TABLE_ID"
        }
      ]
    }
  },
  "direction": "DOWNSTREAM",
  "limits": {
    "maxDepth": 10,
    "maxResults": 5000
  }
}

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

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

{
  "links": [
    {
      "source": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.source_table"
      },
      "target": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.target_table"
      },
      "depth": 1,
      "location": "us"
    }
  ]
}

搜尋多個地理位置

您可以修改 locations 重複陣列欄位中傳遞的地理區域，限制或擴大譜系圖掃描範圍。

C#

C#

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

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

using Google.Api.Gax.Grpc;
using Google.Api.Gax.ResourceNames;
using Google.Cloud.DataCatalog.Lineage.V1;
using System.Threading.Tasks;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for SearchLineageStreaming</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 async Task SearchLineageStreamingRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        SearchLineageStreamingRequest request = new SearchLineageStreamingRequest
        {
            ParentAsLocationName = LocationName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
            Locations = { "", },
            RootCriteria = new SearchLineageStreamingRequest.Types.RootCriteria(),
            Direction = SearchLineageStreamingRequest.Types.SearchDirection.Unspecified,
            Filters = new SearchLineageStreamingRequest.Types.SearchFilters(),
            Limits = new SearchLineageStreamingRequest.Types.SearchLimits(),
        };
        // Make the request, returning a streaming response
        using LineageClient.SearchLineageStreamingStream response = lineageClient.SearchLineageStreaming(request);

        // Read streaming responses from server until complete
        // Note that C# 8 code can use await foreach
        AsyncResponseStream<SearchLineageStreamingResponse> responseStream = response.GetResponseStream();
        while (await responseStream.MoveNextAsync())
        {
            SearchLineageStreamingResponse responseItem = responseStream.Current;
            // Do something with streamed response
        }
        // The response stream has completed
    }
}

Java

Java

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

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

import com.google.api.gax.rpc.ServerStream;
import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.LocationName;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingRequest;
import com.google.cloud.datacatalog.lineage.v1.SearchLineageStreamingResponse;
import java.util.ArrayList;

public class AsyncSearchLineageStreaming {

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

  public static void asyncSearchLineageStreaming() 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 (LineageClient lineageClient = LineageClient.create()) {
      SearchLineageStreamingRequest request =
          SearchLineageStreamingRequest.newBuilder()
              .setParent(LocationName.of("[PROJECT]", "[LOCATION]").toString())
              .addAllLocations(new ArrayList<String>())
              .setRootCriteria(SearchLineageStreamingRequest.RootCriteria.newBuilder().build())
              .setFilters(SearchLineageStreamingRequest.SearchFilters.newBuilder().build())
              .setLimits(SearchLineageStreamingRequest.SearchLimits.newBuilder().build())
              .build();
      ServerStream<SearchLineageStreamingResponse> stream =
          lineageClient.searchLineageStreamingCallable().call(request);
      for (SearchLineageStreamingResponse response : stream) {
        // Do something when a response is received.
      }
    }
  }
}

Node.js

Java

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

如要向 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.
 * TODO(developer): Uncomment these variables before running the sample.
 */
/**
 *  Required. The project and location to initiate the search from.
 */
// const parent = 'abc123'
/**
 *  Required. The locations to search in.
 */
// const locations = ['abc','def']
/**
 *  Required. Criteria for the root of the search.
 */
// const rootCriteria = {}
/**
 *  Required. Direction of the search.
 */
// const direction = {}
/**
 *  Optional. Filters for the search.
 */
// const filters = {}
/**
 *  Optional. Limits for the search.
 */
// const limits = {}

// Imports the Lineage library
const {LineageClient} = require('@google-cloud/lineage').v1;

// Instantiates a client
const lineageClient = new LineageClient();

async function callSearchLineageStreaming() {
  // Construct request
  const request = {
    parent,
    locations,
    rootCriteria,
    direction,
  };

  // Run request
  const stream = await lineageClient.searchLineageStreaming(request);
  stream.on('data', (response) => { console.log(response) });
  stream.on('error', (err) => { throw(err) });
  stream.on('end', () => { /* API call completed */ });
}

callSearchLineageStreaming();

Python

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_v1


def sample_search_lineage_streaming():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.SearchLineageStreamingRequest(
        parent="parent_value",
        locations=["locations_value1", "locations_value2"],
        direction="UPSTREAM",
    )

    # Make the request
    stream = client.search_lineage_streaming(request=request)

    # Handle the response
    for response in stream:
        print(response)

小茹

Ruby

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

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

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the search_lineage_streaming call in the Lineage service
#
# 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/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#search_lineage_streaming.
#
def search_lineage_streaming
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingRequest.new

  # Call the search_lineage_streaming method to start streaming.
  output = client.search_lineage_streaming request

  # The returned object is a streamed enumerable yielding elements of type
  # ::Google::Cloud::DataCatalog::Lineage::V1::SearchLineageStreamingResponse
  output.each do |current_response|
    p current_response
  end
end

REST

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

  • PROJECT_ID：用於管理帳單和配額評估的 Google Cloud 專案 ID。
  • LOCATION_ID： Google Cloud 位置，例如 us-central1。
  • SOURCE_PROJECT_ID：來源資料表所在的 Google Cloud 專案 ID。
  • DATASET_ID：BigQuery 資料集 ID。
  • TABLE_ID：BigQuery 資料表 ID。

HTTP 方法和網址：

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

JSON 要求主體：

{
  "parent": "projects/PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID",
    "us-east1",
    "us-central1"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "bigquery:SOURCE_PROJECT_ID.DATASET_ID.TABLE_ID"
        }
      ]
    }
  },
  "direction": "DOWNSTREAM",
  "limits": {
    "maxDepth": 10,
    "maxResults": 5000
  }
}

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

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

{
  "links": [
    {
      "source": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.source_table"
      },
      "target": {
        "fullyQualifiedName": "bigquery:project-prod.dataset.target_table"
      },
      "depth": 1,
      "location": "us"
    }
  ]
}

根據預設，API 會省略程序資訊 (maxProcessPerLink 預設為 0)。如要擷取建立資料連結的管道資源名稱，請將 limits.maxProcessPerLink 設為非零的正整數。

例如：

Java

Java

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

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

import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.Process;
import com.google.cloud.datacatalog.lineage.v1.ProcessName;

public class SyncGetProcessProcessname {

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

  public static void syncGetProcessProcessname() 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 (LineageClient lineageClient = LineageClient.create()) {
      ProcessName name = ProcessName.of("[PROJECT]", "[LOCATION]", "[PROCESS]");
      Process response = lineageClient.getProcess(name);
    }
  }
}

REST

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

  • BILLING_PROJECT_ID：用於管理帳單和配額評估的 Google Cloud專案。提供專案編號或專案 ID。
  • LOCATION_ID：執行歷程搜尋的 Google Cloud 位置。
  • FULLY_QUALIFIED_NAME：目標實體的完整名稱 (FQN)，格式為 bigquery:PROJECT_ID.DATASET_ID.TABLE_ID。

HTTP 方法和網址：

POST https://datalineage.googleapis.com/v1/projects/BILLING_PROJECT_ID/locations/LOCATION_ID:searchLineageStreaming

JSON 要求主體：

{
  "parent": "projects/BILLING_PROJECT_ID/locations/LOCATION_ID",
  "locations": [
    "LOCATION_ID"
  ],
  "rootCriteria": {
    "entities": {
      "entities": [
        {
          "fullyQualifiedName": "FULLY_QUALIFIED_NAME"
        }
      ]
    }
  },
  "direction": "UPSTREAM",
  "limits": {
    "maxProcessPerLink": 5
  }
}

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

 

回應行為：產生的串流會使用僅包含絕對系統資源名稱 (例如 projects/my-project/locations/us/processes/my-process) 的處理訊息，填入 links[].processes 欄位。

使用 FieldMask 擷取完整程序詳細資料

如需管道的完整結構中繼資料 (例如 displayName、系統 attributes 或執行 origin)，而非僅需資源名稱，則必須使用 API FieldMask：

  1. 請為 limits.maxProcessPerLink 提供非零值。

  2. 在網址路徑中附加 fields 查詢參數，指定 links.processes.process 和其他必要欄位。

例如：

curl -H "Authorization: Bearer ${ACCESS_TOKEN}" \
-H "Content-Type: application/json" \
-X POST "https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming?fields=links.processes.process,links.source,links.target,links.depth" \
--data '{
  "parent": "projects/my-billing-project/locations/us",
  "locations": ["us"],
  "rootCriteria": {
    "entities": {
      "entities": [{
        "fullyQualifiedName": "bigquery:my-project.dataset.target_table"
      }]
    }
  },
  "direction": "UPSTREAM",
  "limits": {
    "maxProcessPerLink": 5
  }
}'

搜尋資料表層級和資料欄層級歷程

您可以在 rootCriteria.entities.entities 清單中提供多個實體，在單一要求中同時搜尋資料表層級 (資產層級) 和資料欄層級 (欄位層級) 的歷程：

  • 如要查看資料表層級歷程，請省略 field 陣列。

  • 如要查看資料欄層級歷程，請在 field 陣列中指定單一資料欄。

例如：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [
             {
               "fullyQualifiedName": "bigquery:my-project.dataset.table_a"
             },
             {
               "fullyQualifiedName": "bigquery:my-project.dataset.table_b",
               "field": ["email"]
             }
           ]
         }
       },
       "direction": "DOWNSTREAM"
     }'

使用萬用字元建立資料欄層級歷程

如要搜尋特定資料表的所有可用資料欄層級歷程，而不逐一列出每個資料欄，請使用萬用字元 * 做為 field 陣列中的單一值。

例如：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table",
             "field": ["*"]
           }]
         }
       },
       "direction": "DOWNSTREAM"
     }'

篩選歷程結果

您可以在要求主體中使用 filters 區塊，縮小歷程搜尋結果範圍。

依依附關係類型篩選

如要將結果限制為特定依附元件類型，例如直接複製 (EXACT_COPY) 或篩選和分組等轉換 (OTHER)，請使用 dependencyTypes 篩選器。

例如：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "dependencyTypes": ["EXACT_COPY"]
       }
     }'

如要確保搜尋結果只傳回資料表層級歷程，並完全排除資料欄層級歷程，請將 entitySet 篩選器設為 ENTITIES。

例如：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "entitySet": "ENTITIES"
       }
     }'

依日期範圍篩選

您可以將歷程搜尋結果限制在特定時間間隔內。

舉例來說，如要搜尋在特定時間戳記後建立的歷程資料，請使用下列要求：

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json" \
     -X POST https://datalineage.googleapis.com/v1/projects/my-billing-project/locations/us:searchLineageStreaming \
     --data '{
       "parent": "projects/my-billing-project/locations/us",
       "locations": ["us"],
       "rootCriteria": {
         "entities": {
           "entities": [{
             "fullyQualifiedName": "bigquery:my-project.dataset.my_table"
           }]
         }
       },
       "direction": "DOWNSTREAM",
       "filters": {
         "timeRange": {
           "startTime": "2026-01-01T00:00:00Z"
         }
       }
     }'

疑難排解：處理無法連線的位置和部分圖表

由於串流 API 會同時掃描分散的一組專案和位置，因此在執行期間，某些遠端區域可能會暫時關閉、無法通訊或設定錯誤。

  • 症狀：傳回的歷程圖版面配置不完整，或缺少預期的區域躍點。

  • 診斷：為保護資料完整性，searchLineageStreamingResponse 串流會使用 projects/PROJECT_NUMBER/locations/LOCATION 格式 (例如 projects/123456789/locations/us-east1)，在專屬的 unreachable 欄位 (重複字串) 中填入有問題的位置。

  • 最佳做法：建構用戶端應用程式時，請務必檢查 unreachable 欄位，確認資料完整性，再處理圖表。

後續步驟