데이터 계보를 확인하여 프로젝트 리소스와 이를 만든 프로세스 간의 관계를 파악합니다. 이러한 관계는 테이블, 데이터 세트와 같은 데이터 애셋이 쿼리, 파이프라인과 같은 프로세스에 의해 어떻게 변환되는지 보여줍니다. 이 가이드에서는 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 카탈로그 뷰어 (
roles/dataplex.catalogViewer)
역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.
이러한 사전 정의된 역할에는 데이터 계보를 보는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.
필수 권한
데이터 계보를 보려면 다음 권한이 필요합니다.
-
BigQuery 테이블 세부정보를 확인합니다.
bigquery.tables.get- 테이블의 스토리지 프로젝트 -
BigQuery 작업 세부정보 보기:
bigquery.jobs.get- 작업의 컴퓨팅 프로젝트
커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.
데이터 계보 뷰 유형
Google Cloud 콘솔에서 계보 정보를 대화형 그래프 또는 구조화된 목록으로 볼 수 있습니다.
그래프 요소 (예: 노드, 가장자리, 프로세스 아이콘, 라벨) 및 목록 뷰에서 사용할 수 있는 열에 관한 자세한 설명은 Knowledge Catalog의 데이터 계보 시각화 정보를 참고하세요.
데이터 계보 사용 설정
데이터 계보를 사용 설정하여 지원되는 시스템의 계보 정보를 자동으로 추적합니다. 기본적으로 API를 사용 설정하면 지원되는 대부분의 서비스에 대해 계보 추적이 활성화됩니다. Managed Service for Apache Spark 계보 수집을 제어하려면 서비스의 계보 수집 제어를 참고하세요.
Data Lineage API는 Knowledge Catalog Premium Processing SKU에 따라 청구됩니다. 자세한 내용은 Knowledge Catalog 가격 책정을 참고하세요.
계보가 표시되는 프로젝트와 계보가 기록되는 프로젝트 모두에서 Data Lineage API를 사용 설정해야 합니다. 자세한 내용은 프로젝트 유형을 참고하세요.
- 계보 정보를 캡처하려면 다음 단계를 완료합니다.
-
Google Cloud 콘솔의 프로젝트 선택기 페이지에서 계보를 기록할 프로젝트를 선택합니다.
Data Lineage API를 사용 설정합니다.
- 계보를 기록하려는 각 프로젝트에 대해 이전 단계를 반복합니다.
-
계보를 보는 프로젝트에서 Data Lineage API 및 Dataplex API를 사용 설정합니다.
서비스의 계보 수집 제어
프로젝트, 폴더 또는 조직 수준에서 특정 서비스에 대해 자동 계보 추적을 선택적으로 사용 설정하거나 중지할 수 있습니다.
리소스 트리를 통해 이러한 구성이 계층적으로 적용되는 방식에 관한 자세한 내용은 계보 수집 제어를 참고하세요.
계보 보기
데이터가 시스템 전반에서 변환되고 이동하는 방식을 추적하려면 Google Cloud 콘솔 또는 API를 사용하여 데이터 계보를 확인하세요.
콘솔
다양한 시작점에서 Google Cloud 콘솔의 데이터 계보 정보에 액세스할 수 있습니다.
- Knowledge Catalog: Knowledge Catalog 검색 페이지로 이동하여 검색 모드로 Knowledge Catalog를 선택하고, 보려는 항목을 검색한 다음 클릭합니다. 자세한 내용은 Knowledge Catalog에서 리소스 검색을 참고하세요.
- BigQuery: BigQuery 페이지로 이동하여 데이터 계보를 확인할 테이블을 엽니다.
- Vertex AI: 데이터 세트 또는 모델 레지스트리 페이지로 이동하여 데이터 계보를 보려는 데이터 세트 또는 모델을 클릭합니다.
계보 그래프를 보려면 다음 단계를 따르세요.
계보 탭을 클릭합니다.
기본 그래프 뷰가 열리고 시스템과 리전 전반의 테이블 수준 계보가 표시됩니다. 자세한 내용은 계보 그래프 뷰를 참조하세요.
계보 그래프를 수동으로 탐색하려면 노드 옆에 있는 펼치기를 클릭하여 한 번에 노드 5개를 더 로드합니다.
자세한 내용은 계보 그래프 수동 탐색을 참조하세요.
그래프 뷰에서 노드를 클릭합니다.
세부정보 패널이 열리고 정규화된 이름 및 유형과 같은 애셋에 대한 정보가 표시됩니다. 자세한 내용은 노드 세부정보를 참조하세요.
그래프 뷰에서 프로세스 아이콘이 있는 에지를 클릭합니다.
쿼리 패널이 열립니다. 자세한 내용은 변환 로직 검사 및 실행 감사 및 기록을 참조하세요.
- 변환 로직을 검사하려면 세부정보 탭을 클릭합니다.
- 실행 감사 및 기록을 보려면 실행 탭을 클릭합니다.
계보 탐색기 패널에서 필터 기준(예: 방향, 종속 항목 유형 또는 기간)을 선택한 후 적용을 클릭합니다.
그러면 특정 리전 내에 포커스가 지정된 뷰가 열립니다(프리뷰). 이 뷰는 최대 3단계 노드까지 그래프를 자동으로 펼칩니다. 자세한 내용은 포커스가 지정된 계보 뷰를 위해 필터 적용을 참조하세요.
포커스가 지정된 그래프 뷰에서 노드를 선택한 후 노드 세부정보 패널에서 경로 시각화를 클릭하여 선택한 노드에서 루트 항목까지의 계보 경로를 시각화합니다(포커스가 지정된 뷰만 해당).
자세한 내용은 계보 경로 시각화를 참조하세요.
열 수준 계보 (BigQuery 및 Managed Service for Apache Spark 작업에만 해당)를 보려면 다음 중 하나를 수행합니다.
- 포커스가 지정된 그래프 뷰의 테이블에서 열 아이콘을 클릭합니다.
열 아이콘 - 계보 탐색기 패널에서 열 이름별로 필터링하고 적용을 클릭합니다.
자세한 내용은 열 수준 계보를 참조하세요.
- 포커스가 지정된 그래프 뷰의 테이블에서 열 아이콘을 클릭합니다.
재설정을 클릭합니다.
이 작업은 적용된 모든 필터를 삭제하고 그래프 뷰 시작 부분으로 이동합니다.
목록을 클릭하여 목록 보기로 전환합니다.
목록 보기는 테이블 수준 및 열 수준 계보의 간소화된 상세 표 형식 표현을 제공하며 그래프 뷰와 동기화됩니다. 기본적으로 간소화된 목록 보기가 표시되며 개별 소스-대상 관계를 분석하기 위해 상세 목록 보기로 전환할 수 있습니다. 표시되는 열을 구성하고 계보 데이터를 내보낼 수 있습니다. 자세한 내용은 계보 목록 보기를 참조하세요.
자바
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}")
계보 시각화 세부 조정
계보 시각화를 세부적으로 조정하려면 계보 탐색기에서 강조 표시 및 필터링 옵션을 사용하면 됩니다.
특정 프로젝트, 데이터 세트 또는 항목 이름을 검색하려면 필터 패널을 사용합니다.
필터를 적용하면 필터 기준과 일치하는 계보 노드가 일치하는 노드로 간주됩니다. 일치하는 노드와 일치하지 않는 노드가 표시되는 방식을 조정할 수 있습니다.
계보 그래프에서 필터 지우기 버튼 옆에 있는 작업 더보기 아이콘을 클릭하여 표시 옵션을 확인합니다.
다음 옵션 중 하나 또는 둘 다를 선택합니다.
두 옵션을 동시에 선택할 수 있습니다. 두 옵션을 모두 선택하면 필터링되지 않은 노드는 숨겨지고 일치하는 노드는 필터링된 그래프 뷰에서 강조 표시됩니다.
데이터 계보 사용 중지
계보 추적을 중지하고 데이터 계보 요금을 방지하려면 데이터 계보 API (datalineage.googleapis.com)가 사용 설정된 각 프로젝트에서 이를 사용 중지하세요.
Dataplex API를 사용 중지해도 데이터 계보가 사용 중지되거나 요금이 청구되지 않는 것은 아닙니다. Data Lineage API를 사용 중지해야 합니다.
데이터 계보를 사용 중지하려면 다음 탭 중 하나를 선택하고 계보 추적이 사용 설정된 각 프로젝트의 단계를 완료하세요.
콘솔
Google Cloud 콘솔에서 사용 설정된 API 및 서비스 페이지로 이동합니다.
API 목록에서 Data Lineage API를 클릭합니다.
API 사용 중지를 클릭합니다.
Data Lineage API 세부정보 페이지에서 API 사용 중지 버튼을 사용합니다. 메시지가 표시되면 사용 중지를 클릭합니다.
gcloud
Data Lineage API를 사용 중지하려면 gcloud services disable 명령어를 사용합니다.
gcloud services disable datalineage.googleapis.com --project=PROJECT_ID
다음을 바꿉니다.
PROJECT_ID: Google Cloud 프로젝트 ID
API를 완전히 사용 중지하지 않고 특정 서비스의 계보 추적을 중지하려면 서비스의 계보 수집 제어를 참고하세요.
다음 단계
- BigQuery 테이블 복사 및 쿼리 작업의 데이터 계보 추적
- 데이터 계보 정보 모델 알아보기
- 데이터 계보 고려사항 및 제한사항에 대해 알아봅니다.
- 데이터 계보 감사 로깅 알아보기
- 데이터 계보 문제 해결 방법 알아보기
- OpenLineage와 통합하는 방법 알아보기
- Managed Service for Apache Spark에서 데이터 계보를 사용하는 방법을 알아보세요.