Daten-Lineage für Google Cloud-Systeme ansehen

Mit der Datenherkunft können Sie die Beziehungen zwischen den Ressourcen Ihres Projekts und den Prozessen, mit denen sie erstellt wurden, nachvollziehen. Diese Beziehungen zeigen, wie Daten-Assets wie Tabellen und Datasets durch Prozesse wie Abfragen und Pipelines transformiert werden. In dieser Anleitung wird beschrieben, wie Sie Details zur Datenherkunft in der Google Cloud Console ansehen oder mit der Data Lineage API abrufen.

Rollen und Berechtigungen

Die Datenherkunft wird automatisch erfasst, wenn Sie die Data Lineage API aktivieren. Sie benötigen keine Administrator- oder Bearbeiterrollen, um den Datenursprung für Ihre Daten-Assets zu erfassen.

Zum Aufrufen des Datenursprungs benötigen Sie bestimmte IAM-Berechtigungen (Identity and Access Management). Abstammungsinformationen werden projektübergreifend erfasst. Sie benötigen also Berechtigungen für mehrere Projekte.

  • Wenn Sie die Data Lineage in Knowledge Catalog, BigQuery oder Vertex AI ansehen, benötigen Sie Berechtigungen zum Aufrufen von Data Lineage-Informationen in dem Projekt, in dem Sie sie ansehen.

  • Wenn Sie die Herkunft von Daten ansehen, die in anderen Projekten aufgezeichnet wurden, benötigen Sie Berechtigungen, um Herkunftsinformationen in den Projekten aufzurufen, in denen sie aufgezeichnet wurden.

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, damit Sie die nötigen Berechtigungen zum Aufrufen der Datenherkunft haben:

  • 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-Datenbetrachter (roles/bigquery.dataViewer) 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 ansehen: Dataplex Catalog Viewer (roles/dataplex.catalogViewer) für das Projekt, in dem Katalogeinträge gespeichert sind

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 des Datenursprungs erforderlich sind. Maximieren Sie den Abschnitt Erforderliche Berechtigungen, um die notwendigen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind erforderlich, um den Datenursprung aufzurufen:

  • So rufen Sie Details zu einer BigQuery-Tabelle auf: bigquery.tables.get – das Speicherprojekt der Tabelle
  • BigQuery-Jobdetails ansehen: bigquery.jobs.get – das Compute-Projekt des Jobs

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

Arten von Ansichten zur Datenherkunft

Sie können Herkunftsinformationen in der Google Cloud Console als interaktives Diagramm oder als strukturierte Liste ansehen.

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

Lineage aktivieren

Aktivieren Sie die Datenherkunft, um automatisch Informationen zur Herkunft für unterstützte Systeme zu erfassen. Wenn Sie die API aktivieren, wird die Herkunftserfassung für die meisten unterstützten Dienste standardmäßig aktiviert. Informationen zum Steuern der Lineage-Erfassung für Managed Service for Apache Spark finden Sie unter Lineage-Erfassung für einen Dienst steuern.

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

  1. So erfassen Sie Informationen zur Herkunft:

    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 den Datenursprung erfassen möchten.

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

    APIs aktivieren

Erfassung von Lineage für einen Dienst steuern

Sie können die automatische Lineage-Erfassung 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 Lineage-Erfassung steuern.

Vorbereitung

Wenn Sie die Aufnahme von Lineage-Daten steuern möchten, müssen Sie die Data Lineage API verwenden. Achten Sie darauf, dass ein Clientprojekt für die Abrechnung und das Kontingent konfiguriert ist, da die Data Lineage API eine clientbasierte API ist.

  1. Aktivieren Sie die datalineage.googleapis.com API in Ihrem Clientprojekt. Weitere Informationen finden Sie unter Datenherkunft aktivieren.

  2. Legen Sie das Kundenprojekt fest. Verwenden Sie für die folgenden Beispiele den Header X-Goog-User-Project. Weitere Informationen finden Sie unter Systemparameter.

Aktuelle Konfiguration abrufen

Wenn Sie prüfen möchten, ob die Aufnahme von Herkunftsinformationen für eine Ressource aktiviert ist, oder den etag-Wert abrufen möchten, bevor Sie die Konfiguration ändern, rufen Sie die aktuelle Konfiguration ab.

C#

C#

Folgen Sie der Einrichtungsanleitung für C# in der Data Lineage-Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel ausprobieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Data Lineage C# API.

Richten Sie zur Authentifizierung bei Data Lineage die Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 

using Google.Cloud.DataCatalog.Lineage.ConfigManagement.V1;

public sealed partial class GeneratedConfigManagementServiceClientSnippets
{
    /// <summary>Snippet for GetConfig</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 void GetConfigRequestObject()
    {
        // Create client
        ConfigManagementServiceClient configManagementServiceClient = ConfigManagementServiceClient.Create();
        // Initialize request argument(s)
        GetConfigRequest request = new GetConfigRequest
        {
            ConfigName = ConfigName.FromProjectLocation("[PROJECT]", "[LOCATION]"),
        };
        // Make the request
        Config response = configManagementServiceClient.GetConfig(request);
    }
}

Go

Go

Folgen Sie der Einrichtungsanleitung für Go in der Data Lineage-Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel ausprobieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Data Lineage Go API.

Richten Sie zur Authentifizierung bei Data Lineage die Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 


//go:build examples

package main

import (
	"context"

	configmanagement "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1"
	configmanagementpb "cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb"
)

func main() {
	ctx := context.Background()
	// 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://pkg.go.dev/cloud.google.com/go#hdr-Client_Options
	c, err := configmanagement.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

	req := &configmanagementpb.GetConfigRequest{
		// TODO: Fill request struct fields.
		// See https://pkg.go.dev/cloud.google.com/go/datacatalog/lineage/configmanagement/apiv1/configmanagementpb#GetConfigRequest.
	}
	resp, err := c.GetConfig(ctx, req)
	if err != nil {
		// TODO: Handle error.
	}
	// TODO: Use resp.
	_ = resp
}

Java

Java

Folgen Sie der Einrichtungsanleitung für Java in der Data Lineage-Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel ausprobieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Data Lineage Java API.

Richten Sie zur Authentifizierung bei Data Lineage die Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 

import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;

public class SyncGetConfig {

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

  public static void syncGetConfig() 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 (ConfigManagementServiceClient configManagementServiceClient =
        ConfigManagementServiceClient.create()) {
      GetConfigRequest request =
          GetConfigRequest.newBuilder()
              .setName(ConfigName.ofProjectLocationName("[PROJECT]", "[LOCATION]").toString())
              .build();
      Config response = configManagementServiceClient.getConfig(request);
    }
  }
}

Python

Python

Folgen Sie der Einrichtungsanleitung für Python in der Data Lineage-Kurzanleitung zur Verwendung von Clientbibliotheken, bevor Sie dieses Beispiel ausprobieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Data Lineage Python API.

Richten Sie zur Authentifizierung bei Data Lineage die Standardanmeldedaten für Anwendungen ein. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 

# 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_configmanagement_v1


def sample_get_config():
    # Create a client
    client = datacatalog_lineage_configmanagement_v1.ConfigManagementServiceClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_configmanagement_v1.GetConfigRequest(
        name="name_value",
    )

    # Make the request
    response = client.get_config(request=request)

    # Handle the response
    print(response)

gcloud

Verwenden Sie den Befehl gcloud datalineage config describe, um die aktuelle Konfiguration der Herkunft anzuzeigen. Sie können die Konfiguration für ein Projekt, einen Ordner oder eine Organisation abrufen.

Das folgende Beispiel zeigt, wie die Konfiguration für das aktuelle Projekt abgerufen wird:

gcloud datalineage config describe

Wenn Sie beispielsweise die Konfiguration für ein bestimmtes Projekt abrufen möchten, verwenden Sie das Flag --project:

gcloud datalineage config describe --project=PROJECT_ID

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aufrufen möchten.

Wenn Sie die aktuelle Konfiguration für die Aufnahme von Lineage-Informationen für einen Dienst für einen Ordner oder eine Organisation aufrufen möchten, ersetzen Sie --project=PROJECT_ID durch:

  • --folder=FOLDER_ID, wenn Sie die Einstellungen für die Datenaufnahme für einen Ordner ansehen möchten.
  • --organization=ORGANIZATION_ID, wenn Sie die Einstellungen für die Datenaufnahme für eine Organisation aufrufen möchten.

REST

Verwenden Sie die Methode projects.locations.config.get, um die aktuelle Konfiguration der Datenherkunft aufzurufen. Sie können die Konfiguration für ein Projekt, einen Ordner oder eine Organisation abrufen.

Das folgende Beispiel zeigt, wie die Konfiguration für ein Projekt abgerufen wird:

Ersetzen Sie diese Werte in den folgenden Anfragedaten:

  • CLIENT_PROJECT_ID: Die ID Ihres Clientprojekts, das für die Abrechnung oder Kontingente verwendet wird.
  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aufrufen möchten.

HTTP-Methode und URL: 

GET https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

Wenn Sie die Anfrage senden möchten, maximieren Sie eine der folgenden Optionen:

Der Befehl gibt eine der folgenden Ausgaben zurück:

  • Wenn Sie keine Einstellungen für die Aufnahme von Lineage-Informationen angeben, erhalten Sie eine Ausgabe mit einem leeren ingestion-Objekt: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {}
}

    Das bedeutet, dass für den Dienst die Standardeinstellung für die Aufnahme von Lineage-Informationen verwendet wird. In diesem Beispiel ist die Einstellung für die Lineage-Erfassung für Managed Service for Apache Spark enabled.

  • Wenn Sie die Lineage-Erfassung explizit aktivieren, erhalten Sie die folgende Ausgabe: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}
  • Wenn die Aufnahme von Lineage deaktiviert ist, wird die folgende Ausgabe angezeigt: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Um die Konfiguration für einen Ordner oder eine Organisation abzurufen, ersetzen Sie projects/PROJECT_ID durch folders/FOLDER_ID oder organizations/ORGANIZATION_ID.

Das Feld etag in der Antwort ist eine Prüfsumme, die vom Server auf Grundlage des aktuellen Werts der Konfiguration generiert wird. Wenn Sie eine Konfiguration mit der Methode patch aktualisieren, können Sie den Wert etag, der von einer aktuellen get-Anfrage zurückgegeben wurde, in den Anfragetext aufnehmen. Wenn Sie etag angeben, wird damit überprüft, ob sich die Konfiguration seit Ihrer letzten Leseanfrage geändert hat. Wenn es eine Diskrepanz gibt, schlägt die Aktualisierungsanfrage fehl. So wird verhindert, dass Sie in Szenarien mit Lese-, Änderungs- und Schreibvorgängen versehentlich Konfigurationen überschreiben, die von anderen Nutzern vorgenommen wurden. Wenn Sie in Ihrer patch-Anfrage keine etag angeben, überschreibt Knowledge Catalog die Konfiguration bedingungslos.

Erfassung von Herkunftsdaten für einen Dienst deaktivieren

Wenn Sie Kosten verwalten, Richtlinien zur Datenverwaltung erzwingen oder Entwicklungsprojekte und andere Arbeitslasten ausschließen möchten, die nicht von der Herkunftsnachverfolgung profitieren, können Sie die Erfassung von Herkunftsdaten für einen Dienst deaktivieren.

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class DisableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    disableLineageIngestion(projectId, location);
  }

  // Disables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void disableLineageIngestion(String projectId, String location) throws Exception {
    // 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 (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to disable.
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to false to disable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(false).build();

      // Build the ingestion rule.
      IngestionRule disableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(disableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def disable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Disables lineage ingestion for a specific service.

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to disable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to False to disable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=False
    )

    # Create the ingestion rule.
    disable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )
     # Append the new disabling rule and assign it back to the config ingestion rules
    new_rules.append(disable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

Wenn Sie die Aufnahme von Lineage-Informationen für einen bestimmten Dienst deaktivieren möchten, verwenden Sie den Befehl gcloud datalineage config update mit einem Inline-JSON-String oder einem Pfad zu einer JSON-Datei, in der lineageEnablement.enabled für den jeweiligen integration auf false festgelegt ist.

Das folgende Beispiel zeigt, wie Sie die Aufnahme von Lineage-Daten für einen Dienst für ein Projekt mithilfe eines Inline-JSON-Strings deaktivieren:

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": false
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aktualisieren möchten.
  • INTEGRATION: Die Integration, für die Sie die Konfiguration festlegen. Beispiel: DATAPROC oder BIGQUERY
  • ETAG: Der etag-Wert, der von einer aktuellen get-Anfrage im Anfragebody zurückgegeben wird. Er wird verwendet, um zu prüfen, ob sich die Konfiguration seit Ihrer letzten Leseanfrage geändert hat.

So aktualisieren Sie die Konfiguration mit einer JSON-Datei:

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

Ersetzen Sie Folgendes:

  • CONFIG_FILE: Der Pfad zur JSON-Datei mit der Konfiguration.

Wenn Sie die Aufnahme von Lineage-Informationen für einen Dienst für einen Ordner oder eine Organisation deaktivieren möchten, ersetzen Sie --project=PROJECT_ID durch:

  • --folder=FOLDER_ID, wenn Sie die Einstellungen für die Datenaufnahme für einen Ordner aktualisieren möchten.
  • --organization=ORGANIZATION_ID, wenn Sie die Einstellungen für die Datenaufnahme für eine Organisation aktualisieren möchten.

REST

Wenn Sie die Aufnahme von Lineage-Informationen für einen bestimmten Dienst deaktivieren möchten, verwenden Sie die Methode projects.locations.config.patch mit einer Aufnahmeregel, die lineageEnablement.enabled für den jeweiligen integration auf false setzt.

Um zu verhindern, dass Konfigurationen, die von anderen Nutzern in Lese-/Änderungs-/Schreibvorgängen vorgenommen wurden, unbeabsichtigt überschrieben werden, können Sie das Feld etag in den Anfragetext aufnehmen. Weitere Informationen finden Sie unter Aktuelle Konfiguration abrufen.

Ersetzen Sie diese Werte in den folgenden Anfragedaten:

  • CLIENT_PROJECT_ID: Die ID Ihres Clientprojekts, das für die Abrechnung oder Kontingente verwendet wird.
  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aktualisieren möchten.
  • ETAG: Der etag-Wert, der von einer aktuellen get-Anfrage zurückgegeben wurde.
  • INTEGRATION: Die integration, für die Sie die Konfiguration festgelegt haben. Beispiel: DATAPROC

HTTP-Methode und URL:

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

JSON-Text anfordern:

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "ETAG"
}

Wenn Sie die Anfrage senden möchten, maximieren Sie eine der folgenden Optionen:

Sie sollten eine JSON-Antwort ähnlich wie diese erhalten:

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Wenn Sie die Aufnahme von Lineage-Informationen für einen Ordner oder eine Organisation deaktivieren möchten, ersetzen Sie projects/PROJECT_ID durch folders/FOLDER_ID oder organizations/ORGANIZATION_ID.

Erfassung von Herkunftsdaten für einen Dienst aktivieren

Wenn Sie das Tracking nach dem Deaktivieren fortsetzen oder eine standardmäßig deaktivierte Integration aktivieren möchten, müssen Sie die Aufnahme von Lineage-Daten für einen Dienst aktivieren.

Java

package com.google.cloud.datacatalog.lineage.configmanagement.v1.samples;

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.Config.Ingestion.IngestionRule.LineageEnablement;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigManagementServiceClient;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.ConfigName;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.GetConfigRequest;
import com.google.cloud.datacatalog.lineage.configmanagement.v1.UpdateConfigRequest;

public class EnableLineageIngestion {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String location = "global";
    enableLineageIngestion(projectId, location);
  }

  // Enables lineage ingestion for a specific service
  // (Managed Service for Apache Spark).
  public static void enableLineageIngestion(String projectId, String location) throws Exception {
    // 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 (ConfigManagementServiceClient client = ConfigManagementServiceClient.create()) {
      // Format the resource name.
      String name = ConfigName.ofProjectLocationName(projectId, location).toString();

      Config.Builder configBuilder = Config.newBuilder().setName(name);

      // It is a best practice to read the existing config to preserve other rules
      // and use the etag for optimistic concurrency control.
      try {
        GetConfigRequest getRequest = GetConfigRequest.newBuilder().setName(name).build();
        Config existingConfig = client.getConfig(getRequest);
        configBuilder.mergeFrom(existingConfig);
      } catch (NotFoundException e) {
        // If config doesn't exist, we will proceed by creating a new one.
      }

      // Create an integration selector for the service you want to enable (e.g., DATAPROC).
      IntegrationSelector selector =
          IntegrationSelector.newBuilder().setIntegration(Integration.DATAPROC).build();

      // Set lineage enablement to true to enable tracking.
      LineageEnablement enablement = LineageEnablement.newBuilder().setEnabled(true).build();

      // Build the ingestion rule.
      IngestionRule enableRule =
          IngestionRule.newBuilder()
              .setIntegrationSelector(selector)
              .setLineageEnablement(enablement)
              .build();

      // Preserve existing rules except for the one we are modifying, then add the new rule.
      // We clear the ingestion block out of the configBuilder entirely to reconstruct it.
      Ingestion.Builder ingestionBuilder = Ingestion.newBuilder();
      if (configBuilder.hasIngestion()) {
        for (IngestionRule rule : configBuilder.getIngestion().getRulesList()) {
          // Keep all existing rules EXCEPT the one targeting DATAPROC
          if (rule.getIntegrationSelector().getIntegration() != Integration.DATAPROC) {
            ingestionBuilder.addRules(rule);
          }
        }
      }
      ingestionBuilder.addRules(enableRule);

      // Update the config builder with the reconstructed ingestion settings.
      configBuilder.setIngestion(ingestionBuilder.build());

      // Build the update request.
      UpdateConfigRequest request = UpdateConfigRequest.newBuilder()
          .setConfig(configBuilder.build())
          .build();

      // Update the config.
      Config response = client.updateConfig(request);
      System.out.printf("Successfully updated config: %s\n", response.getName());
    }
  }
}

Python

from google.api_core.exceptions import NotFound
from google.cloud.datacatalog.lineage import configmanagement_v1

def enable_lineage_ingestion(project_id: str, location: str = "global") -> configmanagement_v1.Config:
    """Enables lineage ingestion for a specific service like Dataproc
    (Managed Service for Apache Spark).

    Args:
        project_id: The ID of your Google Cloud project.
        location: The region location, usually 'global'.

    Returns:
        The updated Configuration object.
    """
    # Initialize client that will be used to send requests.
    client = configmanagement_v1.ConfigManagementServiceClient()

    # The config name format
    name = f"projects/{project_id}/locations/{location}/config"

    try:
        # Retrieve the existing config to preserve other configurations and
        # obtain the latest etag for optimistic concurrency control.
        config = client.get_config(name=name)

        # Filter out existing rules for the integration we are updating
        new_rules = [
            rule for rule in config.ingestion.rules
            if rule.integration_selector.integration != configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
        ]
    except NotFound:
        # If the config does not exist, start fresh
        config = configmanagement_v1.Config(name=name)
        new_rules = []

    # Define the integration to enable tracking for (e.g., DATAPROC).
    integration_selector = configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector(
        integration=configmanagement_v1.Config.Ingestion.IngestionRule.IntegrationSelector.Integration.DATAPROC
    )

    # Set lineage enablement to True to enable tracking.
    lineage_enablement = configmanagement_v1.Config.Ingestion.IngestionRule.LineageEnablement(
        enabled=True
    )

    # Create the ingestion rule.
    enable_rule = configmanagement_v1.Config.Ingestion.IngestionRule(
        integration_selector=integration_selector,
        lineage_enablement=lineage_enablement,
    )

    # Append the new enabling rule and assign it back to the config ingestion rules
    new_rules.append(enable_rule)
    config.ingestion = configmanagement_v1.Config.Ingestion(rules=new_rules)

    # Create the update request using the config (which includes the etag if it existed).
    request = configmanagement_v1.UpdateConfigRequest(
        config=config,
    )

    # Make the request to update the config
    response = client.update_config(request=request)

    print(f"Successfully updated config: {response.name}")
    return response

gcloud

Wenn Sie die Aufnahme von Lineage-Informationen für einen bestimmten Dienst aktivieren möchten, verwenden Sie den Befehl gcloud datalineage config update mit einem Inline-JSON-String oder einem Pfad zu einer JSON-Datei, in der lineageEnablement.enabled für den jeweiligen integration auf true festgelegt ist. Zu den aktuellen Integrationen gehören Managed Service for Apache Spark, BigQuery und Managed Airflow.

Im folgenden Beispiel wird gezeigt, wie Sie die Aufnahme von Lineage-Daten für einen Dienst für ein Projekt mithilfe eines Inline-JSON-Strings aktivieren:

gcloud datalineage config update --project=PROJECT_ID \
  --config='{
    "ingestion": {
      "rules": [
        {
          "integrationSelector": {
            "integration": "INTEGRATION"
          },
          "lineageEnablement": {
            "enabled": true
          }
        }
      ]
    },
    "etag": "ETAG"
  }'

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aktualisieren möchten.
  • INTEGRATION: Die Integration, für die Sie die Konfiguration festlegen (z. B. DATAPROC oder BIGQUERY).
  • ETAG: Der etag-Wert, der von einer aktuellen get-Anfrage im Anfragebody zurückgegeben wird. Er wird verwendet, um zu prüfen, ob sich die Konfiguration seit Ihrer letzten Leseanfrage geändert hat.

So aktualisieren Sie die Konfiguration mit einer JSON-Datei:

gcloud datalineage config update --project=PROJECT_ID --config=CONFIG_FILE

Ersetzen Sie Folgendes:

  • CONFIG_FILE: Der Pfad zur JSON-Datei mit der Konfiguration.

Wenn Sie die Erfassung von Lineage-Informationen für einen Dienst für einen Ordner oder eine Organisation aktivieren möchten, ersetzen Sie --project=PROJECT_ID durch:

  • --folder=FOLDER_ID, wenn Sie die Einstellungen für die Datenaufnahme für einen Ordner aktualisieren möchten.
  • --organization=ORGANIZATION_ID, wenn Sie die Einstellungen für die Datenaufnahme für eine Organisation aktualisieren möchten.

REST

Wenn Sie die Aufnahme von Lineage-Informationen für einen bestimmten Dienst aktivieren möchten, verwenden Sie die Methode projects.locations.config.patch mit einer Aufnahmeregel, die lineageEnablement.enabled für die jeweilige integration auf true festlegt. Zu den aktuellen Integrationen gehören Managed Service for Apache Spark, BigQuery und Managed Airflow.

Um zu verhindern, dass Konfigurationen, die von anderen Nutzern in Lese-/Änderungs-/Schreibvorgängen vorgenommen wurden, unbeabsichtigt überschrieben werden, können Sie das Feld etag in den Anfragetext aufnehmen. Weitere Informationen finden Sie unter Aktuelle Konfiguration abrufen.

Ersetzen Sie diese Werte in den folgenden Anfragedaten:

  • CLIENT_PROJECT_ID: Die ID Ihres Clientprojekts, das für die Abrechnung oder Kontingente verwendet wird.
  • PROJECT_ID: Die ID des Projekts, dessen Konfiguration Sie aktualisieren möchten.
  • ETAG: Der etag-Wert, der von einer aktuellen get-Anfrage zurückgegeben wurde.
  • INTEGRATION: Die integration, für die Sie die Konfiguration festgelegt haben. Beispiel: DATAPROC

HTTP-Methode und URL:

PATCH https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/global/config

JSON-Text anfordern:

{
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "ETAG"
}

Wenn Sie die Anfrage senden möchten, maximieren Sie eine der folgenden Optionen:

Sie sollten eine JSON-Antwort ähnlich wie diese erhalten:

{
  "name": "projects/PROJECT_ID/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "INTEGRATION"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Um die Aufnahme von Lineage-Informationen für einen Dienst für einen Ordner oder eine Organisation zu aktivieren, ersetzen Sie projects/PROJECT_ID durch folders/FOLDER_ID oder organizations/ORGANIZATION_ID.

Herkunft ansehen

Wenn Sie nachvollziehen möchten, wie Daten transformiert werden und sich in Systemen bewegen, können Sie die Datenherkunft über die Google Cloud Console oder die API ansehen.

Console

Sie können von verschiedenen Startpunkten aus in der Google Cloud -Konsole auf Informationen zum Datenursprung zugreifen:

  • Knowledge Catalog:Rufen Sie die Seite Suchen im Knowledge Catalog auf, wählen Sie Knowledge Catalog als Suchmodus aus, suchen Sie nach dem Eintrag, den Sie aufrufen möchten, und klicken Sie dann darauf. Weitere Informationen finden Sie unter Nach Ressourcen in Knowledge Catalog suchen.
  • BigQuery:Rufen Sie die Seite BigQuery auf und öffnen Sie die Tabelle, für die Sie den Datenursprung sehen möchten.
  • Vertex AI:Rufen Sie die Seite Datasets oder Model Registry auf und klicken Sie auf das Dataset oder Modell, für das Sie den Datenursprung sehen möchten.

So rufen Sie das Herkunftsdiagramm auf:

  1. Klicken Sie auf den Tab Lineage.

    Die Standardansicht Diagramm wird geöffnet. Sie zeigt die Lineage auf Tabellenebene über Systeme und Regionen hinweg. Weitere Informationen finden Sie unter Ansicht „Abstammungsdiagramm“.

  2. Wenn Sie den Lineage-Graphen manuell untersuchen möchten, klicken Sie neben einem Knoten auf Maximieren, um jeweils fünf weitere Knoten zu laden.

    Weitere Informationen finden Sie unter Abstammungsdiagramm manuell untersuchen.

  3. Klicken Sie in der Graphansicht auf einen Knoten.

    Der Bereich Details wird mit Informationen zum Asset geöffnet, z. B. mit dem vollständig qualifizierten Namen und dem Typ. Weitere Informationen finden Sie unter Knotendetails.

  4. Klicken Sie in der Ansicht Graph auf eine Kante mit einem Prozesssymbol.

    Der Bereich Abfrage wird geöffnet. Weitere Informationen finden Sie unter Transformationslogik prüfen und Ausführungen prüfen und Verlauf ansehen.

    • Klicken Sie auf den Tab Details, um die Transformationslogik zu prüfen.
    • Klicken Sie auf den Tab Ausführungen, um den Audit- und Ausführungsverlauf aufzurufen.

  5. Wählen Sie im Bereich Lineage Explorer Filterkriterien aus, z. B. Richtung, Abhängigkeitstyp oder Zeitraum, und klicken Sie dann auf Anwenden.

    Dadurch wird eine fokussierte Ansicht in einer bestimmten Region geöffnet (Vorschau). In dieser Ansicht wird das Diagramm automatisch auf bis zu drei Knotenebenen erweitert. Weitere Informationen finden Sie unter Filter anwenden, um eine fokussierte Lineage-Ansicht zu erhalten.

  6. Wählen Sie in der fokussierten Graph-Ansicht einen Knoten aus und klicken Sie dann im Detailbereich des Knotens auf Pfad visualisieren, um den Lineage-Pfad vom ausgewählten Knoten zurück zum Stammknoten zu visualisieren (nur in der fokussierten Ansicht).

    Weitere Informationen finden Sie unter Visualisierung des Lineage-Pfads.

  7. Wenn Sie die Herkunft auf Spaltenebene aufrufen möchten (nur für BigQuery- und Managed Service for Apache Spark-Jobs), haben Sie folgende Möglichkeiten:

    • Klicken Sie in einer fokussierten Graph-Ansicht in einer Tabelle auf das Spaltensymbol.
      Symbol zum Wechseln zur Herkunft auf Spaltenebene.
      Spaltensymbol
    • Filtern Sie im Bereich Lineage Explorer nach Spaltenname und klicken Sie auf Übernehmen.

    Weitere Informationen finden Sie unter Herkunft auf Spaltenebene.

  8. Klicken Sie auf  Zurücksetzen.

    Durch diese Aktion werden alle angewendeten Filter entfernt und Sie gelangen zum Anfang der Diagrammansicht.

  9. Klicken Sie auf Liste, um zur Listenansicht zu wechseln.

    Die Listenansicht bietet vereinfachte und detaillierte tabellarische Darstellungen der Herkunft sowohl auf Tabellen- als auch auf Spaltenebene, die mit der Diagrammansicht synchronisiert werden. Standardmäßig wird die vereinfachte Listenansicht angezeigt. Sie können zur detaillierten Listenansicht wechseln, um einzelne Quell-Ziel-Beziehungen zu analysieren. Sie können konfigurieren, welche Spalten angezeigt werden, und Herkunftsdaten exportieren. Weitere Informationen finden Sie unter Lineage-Listenansicht.

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

Visualisierung des Herkunftspfads optimieren

Um die Lineage-Visualisierung zu optimieren, können Sie im Lineage Explorer Optionen zum Hervorheben und Filtern 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 Lineage-Knoten, die Ihren Filterkriterien entsprechen, als übereinstimmende Knoten betrachtet. Sie können festlegen, wie übereinstimmende und nicht übereinstimmende Knoten dargestellt werden.

  2. Klicken Sie im Lineage-Diagramm neben dem Button Filter löschen auf das Symbol Weitere Aktionen , um Anzeigeoptionen aufzurufen.

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

Optionen zum Hervorheben und Filtern im Lineage Explorer.
Optionen zum Hervorheben und Filtern.

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.

Nächste Schritte