서버 측 자동화를 사용하여 멀티 리전 계보 검색

searchLineageStreaming API를 사용하면 다중 수준의 교차 리전 데이터 계보를 조회하고 통합 그래프를 실시간 스트림으로 수신할 수 있습니다. 요청 제한 시간 없이 광범위하거나 심층적이거나 다중 프로젝트 데이터 아키텍처를 순회하는 도구를 빌드할 수 있습니다. 자세한 내용은 멀티 리전 계보 검색 정보를 참조하세요.

주요 기능

searchLineageStreaming API에는 다음 기능이 포함됩니다.

  • 너비 우선 검색: 정의된 루트 항목 집합에서 업스트림 또는 다운스트림으로 계보 그래프 레이어를 레이어별로 순회하여 연결된 각 애셋의 깊이를 정확하게 계산합니다.

  • 스트리밍 응답: 백엔드 시스템에서 하위 그래프와 계보 링크를 발견하면 이를 포함하는 청크로 분할된 실시간 응답을 제공합니다. 대규모 멀티 프로젝트 그래프에서 타임아웃될 수 있는 표준 계보 조회 API와 달리 스트리밍은 광범위하거나 심층적인 그래프에서 요청 타임아웃을 방지합니다.

  • 다중 위치 및 다중 프로젝트 순회: 요청 경로에 결제 프로젝트를 하나만 지정하더라도 API는 필요한 권한이 있는 경우 여러 Google Cloud 프로젝트 및 지리적 위치에서 계보 링크를 자동으로 검색하고 순회합니다.

  • 세분화된 열 수준 계보: 애셋 간의 열 수준 종속 항목 검색을 지원합니다.

  • 와일드 카드 조회: 정규화된 이름 (FQN)에 *를 추가하여 특정 항목의 모든 열 수준 계보를 검색할 수 있습니다.

  • 파이프라인 통계: 계보 링크를 만든 변환 파이프라인 (프로세스)에 대한 메타데이터를 선택적으로 검색합니다.

시작하기 전에

API에 요청하기 전에 다음 보안 및 환경 기본 요건을 충족하는지 확인하세요.

필요한 역할

데이터 계보 링크를 검색하는 데 필요한 권한을 얻으려면 관리자에게 계보 링크 및 프로세스가 저장된 프로젝트에 대한 데이터 계보 뷰어 (roles/datalineage.viewer) IAM 역할을 부여해 달라고 요청하세요. 역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이 사전 정의된 역할에는 데이터 계보 링크를 검색하는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

데이터 계보 링크를 검색하려면 다음 권한이 필요합니다.

  • 항목 수준 계보 검색: 링크가 저장된 프로젝트에 대한 datalineage.events.get 권한
  • 열 수준 계보 검색: 링크가 저장된 프로젝트에 대한 datalineage.events.getFields 권한
  • 전체 파이프라인 프로세스 세부정보 검색: 프로세스가 저장된 프로젝트에 대한 datalineage.processes.get 권한

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.

리소스 범위 지정

API 요청을 구성할 때는 관리 결제에 사용되는 리소스와 API에서 스캔하는 실제 위치를 구분해야 합니다.

  • 결제 상위 경로: URL 요청의 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# 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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

이 샘플을 사용해 보기 전에 Data Lineage 빠른 시작: 클라이언트 라이브러리 사용의 Java 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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

Ruby

이 샘플을 사용해 보기 전에 Data Lineage 빠른 시작: 클라이언트 라이브러리 사용의 Ruby 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 메서드 및 URL: 

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# 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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

이 샘플을 사용해 보기 전에 Data Lineage 빠른 시작: 클라이언트 라이브러리 사용의 Java 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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

Ruby

이 샘플을 사용해 보기 전에 Data Lineage 빠른 시작: 클라이언트 라이브러리 사용의 Ruby 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 메서드 및 URL: 

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를 0이 아닌 양의 정수로 구성합니다.

예를 들면 다음과 같습니다.

자바

Java

이 샘플을 사용해 보기 전에 Data Lineage 빠른 시작: 클라이언트 라이브러리 사용의 Java 설정 안내를 따르세요. 자세한 내용은 데이터 계보 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 메서드 및 URL: 

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에 0이 아닌 값을 제공하세요.

  2. fields 쿼리 매개변수를 URL 경로에 추가하여 다른 필수 필드와 함께 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 필드를 검사하여 데이터 완전성을 확인하도록 클라이언트 애플리케이션을 빌드하세요.

다음 단계