Daten-Lineage für Google Cloud-Systeme ansehen

Mit der Datenherkunft in Knowledge Catalog (früher Dataplex Universal Catalog) können Sie nachvollziehen, wie sich Daten-Assets durch Systeme bewegen und sich verändern. So können Sie Transformationen prüfen, den Ausführungsverlauf einsehen und Abhängigkeiten analysieren. Nachdem Sie das Tracking aktiviert haben, können Sie Diagramme und Listen für BigQuery, Vertex AI und katalogisierte Einträge in der Google Cloud Console aufrufen oder Herkunftsdetails mit der Data Lineage API abrufen. Weitere Informationen finden Sie unter Informationen zur Datenherkunft.

Rollen und Berechtigungen

Die Datenherkunft wird automatisch erfasst, wenn Sie die Data Lineage API aktivieren. Sie benötigen keine Administrator- oder Bearbeiterrollen, um die Herkunft Ihrer Daten-Assets zu erfassen.

Zum Aufrufen der Datenherkunft benötigen Sie bestimmte IAM-Berechtigungen (Identity and Access Management). Herkunftsinformationen werden projektübergreifend erfasst. Sie benötigen daher Berechtigungen in mehreren Projekten.

• Wenn Sie die Datenherkunft in Knowledge Catalog, BigQuery oder Vertex AI ansehen, benötigen Sie Berechtigungen zum Ansehen von Datenherkunftsinformationen in dem Projekt, in dem Sie sie ansehen.

• Wenn Sie die Herkunft ansehen, die in anderen Projekten aufgezeichnet wurde: Sie benötigen Berechtigungen, um Herkunftsinformationen in den Projekten anzusehen, in denen sie aufgezeichnet wurden.

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Aufrufen der Datenherkunft benötigen:

• Alle: Data Lineage-Betrachter (roles/datalineage.viewer) für das Projekt, in dem die Datenherkunft aufgezeichnet wird, und das Projekt, in dem die Datenherkunft angezeigt wird
• BigQuery-Tabellendetails ansehen: BigQuery Metadata Viewer (roles/bigquery.metadataViewer) für das Speicherprojekt der Tabelle
• BigQuery-Jobdetails ansehen: BigQuery Resource Viewer (roles/bigquery.resourceViewer) für das Compute-Projekt des Jobs
• Details zu anderen katalogisierten Assets aufrufen: Dataplex Catalog Viewer (roles/dataplex.catalogViewer) für das Projekt, in dem Katalogeinträge gespeichert sind
• Dataform-Repositories oder BigQuery-Pipelines aus dem Herkunftsdiagramm öffnen: Dataform-Betrachter (roles/dataform.viewer) für das Projekt, in dem das Dataform-Repository gespeichert ist

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Diese vordefinierten Rollen enthalten die Berechtigungen, die zum Aufrufen der Datenherkunft erforderlich sind. Maximieren Sie den Abschnitt Erforderliche Berechtigungen, um die notwendigen Berechtigungen anzuzeigen:

Sie können diese Berechtigungen auch mit benutzerdefinierten Rollen oder anderen vordefinierten Rollen erhalten.

Arten von Datenherkunftsansichten

Sie können Herkunftsinformationen als interaktives Diagramm oder als strukturierte Liste in der Google Cloud Console ansehen. In beiden Ansichten werden die Beziehungen zwischen den Ressourcen Ihres Projekts und den Prozessen, die sie erstellt oder transformiert haben, angezeigt. Sie können beispielsweise sehen, wie Abfragen und Pipelines Daten-Assets wie Tabellen, Datasets und Modelle transformieren.

Eine detaillierte Beschreibung der Diagrammelemente (z. B. Knoten, Kanten, Prozesssymbole und Labels) und der in den Listenansichten verfügbaren Spalten finden Sie unter Datenherkunftsvisualisierung in Knowledge Catalog.

Datenherkunft aktivieren

Aktivieren Sie die Datenherkunft, um automatisch Herkunftsinformationen für unterstützte Systeme zu erfassen. Wenn Sie die API aktivieren, wird standardmäßig das Herkunfts-Tracking für die meisten unterstützten Dienste aktiviert. Weitere Informationen finden Sie unter Automatisches Datenherkunfts-Tracking und Informationen zur Steuerung der Datenherkunftsaufnahme.

Knowledge Catalog stellt die Datenherkunft über die Premium-Verarbeitungs-SKU und die Metadatenspeicher-SKU in Rechnung. Weitere Informationen finden Sie im Abschnitt Preise unter „Informationen zur Datenherkunft“ und unter Knowledge Catalog – Preise.

Sie müssen die Data Lineage API sowohl in dem Projekt, in dem Sie die Datenherkunft ansehen, als auch in den Projekten, in denen die Datenherkunft aufgezeichnet wird, aktivieren. Weitere Informationen finden Sie unter Projekttypen.

1. So erfassen Sie Informationen zur Datenherkunft:

   1. Wählen Sie in der Google Cloud Console auf der Seite Projektauswahl das Projekt aus, in dem Sie die Herkunft aufzeichnen möchten.

      Zur Projektauswahl

   2. Aktivieren Sie die Data Lineage API.

      API aktivieren

   3. Wiederholen Sie die vorherigen Schritte für jedes Projekt, für das Sie die Herkunft aufzeichnen möchten.

2. Aktivieren Sie in dem Projekt, in dem Sie die Datenherkunft ansehen, die Data Lineage API und die Dataplex API.

   APIs aktivieren

Herkunftsaufnahme für einen Dienst steuern

Sie können das automatische Lineage-Tracking für bestimmte Dienste auf Projekt-, Ordner- oder Organisationsebene selektiv aktivieren oder deaktivieren.

Weitere Informationen dazu, wie diese Konfigurationen hierarchisch über den Ressourcenbaum angewendet werden, finden Sie unter Informationen zur Steuerung der Datenherkunftserfassung.

Herkunft ansehen

Wenn Sie nachvollziehen möchten, wie Daten transformiert und zwischen Systemen verschoben werden, können Sie die Datenherkunft über die Google Cloud Console oder die API aufrufen.

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());
          }
        }
      }
    }
  }
}

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}")

Visualisierung der Herkunft verfeinern

Wenn Sie die Visualisierung der Herkunft verfeinern möchten, können Sie im Herkunfts-Explorer Hervorhebungs- und Filteroptionen verwenden:

1. Wenn Sie nach bestimmten Projekten, Datasets oder Entitätsnamen suchen möchten, verwenden Sie den Bereich Filter.

   Nachdem Sie Filter angewendet haben, werden Herkunftsknoten, die Ihren Filterkriterien entsprechen, als übereinstimmende Knoten betrachtet. Sie können die Anzeige übereinstimmender und nicht übereinstimmender Knoten verfeinern.

2. Klicken Sie im Herkunftsdiagramm auf das Symbol Weitere Aktionen more_vert neben der Schaltfläche Filter löschen, um die Anzeigeoptionen aufzurufen.

3. Wählen Sie eine oder beide der folgenden Optionen aus:

Sie können beide Optionen gleichzeitig auswählen. Wenn beide Optionen ausgewählt sind, werden ungefilterte Knoten ausgeblendet und übereinstimmende Knoten in der gefilterten Diagrammansicht hervorgehoben.

Datenherkunft deaktivieren

Wenn Sie das Tracking neuer Herkunftsereignisse und neue Gebühren für die Premium-Verarbeitung von Datenherkunft (DCU) beenden möchten, deaktivieren Sie die Data Lineage API (datalineage.googleapis.com) in jedem Projekt, in dem sie aktiviert ist.

Durch das Deaktivieren der Dataplex API wird die Datenherkunft nicht deaktiviert und die Gebühren werden nicht beendet. Sie müssen die Data Lineage API deaktivieren. Nachdem Sie die API deaktiviert haben, fallen für zuvor aufgenommene Herkunftsmetadaten bis zu 30 Tage lang Gebühren für die Metadatenspeicherung an, bis die Aufbewahrungsdauer abläuft oder Sie die Metadaten löschen.

Wählen Sie einen der folgenden Tabs aus und führen Sie die Schritte für jedes Projekt aus, in dem die Datenherkunft aktiviert wurde, um die Datenherkunft zu deaktivieren:

gcloud services disable datalineage.googleapis.com --project=PROJECT_ID

Wenn Sie das Lineage-Tracking für bestimmte Dienste beenden möchten, ohne die API vollständig zu deaktivieren, finden Sie weitere Informationen unter Daten-Lineage-Aufnahme für einen Dienst konfigurieren.

Nächste Schritte

• Datenherkunft für eine BigQuery-Tabelle nachverfolgen
• Informationsmodell zur Datenherkunft
• Überlegungen und Einschränkungen zur Datenherkunft
• Informationen zum Audit-Logging der Data Lineage API
• Probleme mit der Datenherkunft beheben.
• In OpenLineage einbinden
• Datenherkunft mit Managed Service for Apache Spark verwenden