借助 Knowledge Catalog（以前称为 Dataplex Universal Catalog）中的数据沿袭功能，您可以跟踪数据资产在系统中的移动和变化情况，以检查转换、查看运行历史记录和分析依赖项。启用跟踪后，您可以在 Google Cloud 控制台中探索 BigQuery、Vertex AI 和已编目条目的图表和列表，也可以使用 Data Lineage API 检索沿袭详细信息。如需了解详情，请参阅数据沿袭简介。
角色与权限
启用 Data Lineage API 后，数据沿袭会自动跟踪沿袭信息。您无需任何管理员或编辑者角色即可捕获数据资产的沿袭。
如需查看数据沿袭，您需要拥有特定的 Identity and Access Management (IAM) 权限。系统会跨项目捕获沿袭信息，因此您需要多个项目的权限。
在 Knowledge Catalog、BigQuery 或 Vertex AI 中查看沿袭时：您需要拥有在查看沿袭信息的项目中查看沿袭信息的权限。
查看在其他项目中记录的沿袭时：您需要拥有在记录沿袭的项目中查看沿袭信息的权限。
如需获得查看数据沿袭所需的权限，请让您的管理员为您授予以下 IAM 角色：
-
全部：针对记录沿袭的项目以及查看沿袭的项目的 Data Lineage Viewer (
roles/datalineage.viewer) -
查看 BigQuery 表详细信息：针对表的存储项目的 BigQuery Metadata Viewer (
roles/bigquery.metadataViewer) -
查看 BigQuery 作业详情：作业计算项目的 BigQuery Resource Viewer (
roles/bigquery.resourceViewer) -
查看其他已编目资产的详细信息：针对存储目录条目的项目的 Dataplex Catalog Viewer (
roles/dataplex.catalogViewer)
如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。
这些预定义角色包含查看数据沿袭所需的权限。如需查看所需的确切权限，请展开所需权限部分：
所需权限
您需要具备以下权限才能查看数据沿袭：
-
查看 BigQuery 表详情：
bigquery.tables.get- 表的存储项目 -
查看 BigQuery 作业详情：
bigquery.jobs.get- 作业的计算项目
数据沿袭视图的类型
您可以在 Google Cloud 控制台中以交互式图表或结构化列表的形式查看沿袭信息。这两个视图都显示了项目资源与创建或转换这些资源的进程之间的关系。例如，您可以查看查询和流水线如何转换数据资产，例如表、数据集和模型。
如需详细了解图表元素（例如节点、边缘、进程图标和标签）以及列表视图中可用的列，请参阅关于 Knowledge Catalog 中的数据沿袭可视化。
启用数据沿袭
启用数据沿袭功能后，系统会开始自动跟踪受支持系统的沿袭信息。默认情况下，启用该 API 会为大多数受支持的服务激活沿袭跟踪。如需了解详情，请参阅自动数据沿袭跟踪和数据沿袭注入控制简介。
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：前往数据集或 Model Registry 页面，然后点击要查看其数据沿袭的数据集或模型。
如需查看沿袭图，请按照以下步骤操作：
点击沿袭标签页。
系统会打开默认的图表视图，其中显示了跨系统和区域的表级沿袭。如需了解详情，请参阅沿袭图视图。
如需手动探索沿袭图，请点击节点旁边的展开，一次加载五个节点。
如需了解详情，请参阅探索沿袭图。
点击图表视图中的节点。
详细信息面板随即打开，其中包含有关资产的信息，例如完全限定名称和类型。如需了解详情，请参阅查看沿袭节点详细信息。
在图表视图中，点击带有进程图标的边缘。
查询面板随即会打开。如需了解详情，请参阅检查转换逻辑和审核和运行历史记录。
- 如需检查转换逻辑，请点击详细信息标签页。
- 如需查看运行的审核和历史记录，请点击运行标签页。
在沿袭探索器面板中，选择过滤条件（例如方向、依赖关系类型或时间范围），然后点击应用。
这会在特定区域内打开一个聚焦视图（预览版）。此视图会自动将图表展开到最多三个级别的节点。如需了解详情，请参阅过滤和突出显示数据沿袭视图。
在重点突出的图表视图中，选择一个节点，然后在该节点的详细信息面板中，点击可视化路径，以直观呈现从所选节点返回到根条目的沿袭路径（仅在重点突出的视图中）。
如需了解详情，请参阅直观呈现数据沿袭路径。
如需查看列级沿袭（仅适用于 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}")
优化沿袭可视化图表
如需优化沿袭可视化，您可以使用沿袭探索器中的突出显示和过滤选项：
如需搜索特定项目、数据集或实体名称，请使用过滤条件面板。
应用过滤条件后，符合过滤条件的沿袭节点会被视为匹配节点。您可以优化匹配和不匹配节点的显示方式。
在沿袭图中，点击清除过滤条件按钮旁边的更多操作 图标，查看显示选项。
选择以下一个或两个选项：
您可以同时选择这两个选项。如果同时选择这两个选项，则未过滤的节点会被隐藏，匹配的节点会在过滤后的图谱视图中突出显示。
停用数据沿袭
如需停止跟踪沿袭并避免产生数据沿袭费用，请在每个启用了 Data Lineage API (datalineage.googleapis.com) 的项目中停用该 API。
停用 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，请参阅为服务配置数据沿袭注入。