Visualizza la tracciabilità dei dati per i sistemi Google Cloud

Visualizza la derivazione dei dati per comprendere le relazioni tra le risorse del tuo progetto e i processi che le hanno create. Queste relazioni mostrano come gli asset di dati, come tabelle e set di dati, vengono trasformati da processi come query e pipeline. Questa guida descrive come visualizzare i dettagli della derivazione dei dati nella console Google Cloud o recuperarli utilizzando l'API Data Lineage.

Ruoli e autorizzazioni

La derivazione dei dati tiene traccia automaticamente delle informazioni sulla derivazione quando abiliti l'API Data Lineage. Non hai bisogno di ruoli di amministratore o editor per acquisire la derivazione degli asset di dati.

Per visualizzare la derivazione dei dati, devi disporre di autorizzazioni Identity and Access Management (IAM) specifiche. Le informazioni sulla derivazione vengono acquisite in più progetti, pertanto devi disporre delle autorizzazioni in più progetti.

  • Quando visualizzi la derivazione in Knowledge Catalog, BigQuery o Vertex AI: devi disporre delle autorizzazioni per visualizzare le informazioni sulla derivazione nel progetto in cui le visualizzi.

  • Quando visualizzi la derivazione registrata in altri progetti, devi disporre delle autorizzazioni per visualizzare le informazioni sulla derivazione nei progetti in cui è stata registrata.

Per ottenere le autorizzazioni necessarie per visualizzare la derivazione dei dati, chiedi all'amministratore di concederti i seguenti ruoli IAM:

  • Visualizzatore Data Lineage (roles/datalineage.viewer) sul progetto in cui viene registrata la derivazione e sul progetto in cui viene visualizzata la derivazione
  • Visualizza i dettagli della tabella BigQuery: Visualizzatore dati BigQuery (roles/bigquery.dataViewer) nel progetto di archiviazione della tabella
  • Visualizza i dettagli del job BigQuery: Visualizzatore risorse BigQuery (roles/bigquery.resourceViewer) nel progetto di computing del job
  • Visualizza i dettagli di altre risorse catalogate: Visualizzatore Dataplex Catalog (roles/dataplex.catalogViewer) sul progetto in cui sono archiviate le voci di catalogo

Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questi ruoli predefiniti contengono le autorizzazioni necessarie per visualizzare la derivazione dei dati. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per visualizzare la derivazione dei dati sono necessarie le seguenti autorizzazioni:

  • Visualizza i dettagli della tabella BigQuery: bigquery.tables.get: il progetto di archiviazione della tabella
  • Visualizza i dettagli del job BigQuery: bigquery.jobs.get: il progetto di computing del job

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Tipi di visualizzazioni della derivazione dei dati

Puoi visualizzare le informazioni sulla derivazione come grafico interattivo o elenco strutturato nella console Google Cloud .

Per una descrizione dettagliata degli elementi del grafico (come nodi, archi, icone di processo ed etichette) e delle colonne disponibili nelle visualizzazioni elenco, consulta Informazioni sulla visualizzazione della tracciabilità dei dati in Knowledge Catalog.

Abilita la derivazione dei dati

Attiva la derivazione dei dati per iniziare a monitorare automaticamente le informazioni sulla derivazione per i sistemi supportati. Per impostazione predefinita, l'abilitazione dell'API attiva il monitoraggio della derivazione per la maggior parte dei servizi supportati. Per controllare l'importazione della derivazione di Managed Service for Apache Spark, consulta Controllare l'importazione della derivazione per un servizio.

Devi abilitare l'API Data Lineage sia nel progetto in cui visualizzi la derivazione sia nei progetti in cui viene registrata. Per saperne di più, consulta Tipi di progetti.

  1. Per acquisire le informazioni sulla derivazione, completa i seguenti passaggi:

    1. Nella console Google Cloud , nella pagina Selettore progetto, seleziona il progetto in cui vuoi registrare la derivazione.

      Vai al selettore di progetti

    2. Abilita l'API Data Lineage.

      Abilita l'API

    3. Ripeti i passaggi precedenti per ogni progetto in cui vuoi registrare la derivazione.

  2. Nel progetto in cui visualizzi la derivazione, abilita l'API Data Lineage e l'API Dataplex.

    Abilita le API

Controllare l'importazione della derivazione per un servizio

Puoi attivare o disattivare in modo selettivo il monitoraggio automatico della derivazione per servizi specifici a livello di progetto, cartella o organizzazione.

Per informazioni dettagliate su come queste configurazioni vengono applicate gerarchicamente tramite l'albero delle risorse, vedi Controllare l'importazione della derivazione.

Prerequisiti

Per controllare l'importazione della derivazione, devi utilizzare l'API Data Lineage. Assicurati di avere un progetto client configurato per la fatturazione e la quota, perché l'API Data Lineage è un'API basata sul client.

  1. Abilita l'API datalineage.googleapis.com nel progetto client. Per maggiori informazioni, vedi Attivare la derivazione dei dati.

  2. Imposta il progetto client. Per gli esempi seguenti, utilizza l'intestazione X-Goog-User-Project. Per ulteriori informazioni, vedi Parametri di sistema.

Ottieni la configurazione attuale

Per verificare se l'importazione della derivazione è abilitata per una risorsa o per ottenere il valore di etag prima di modificare la configurazione, recupera la configurazione corrente.

C#

C#

Prima di provare questo esempio, segui le istruzioni di configurazione di C# nella guida rapida di Data Lineage per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Data Lineage C#.

Per eseguire l'autenticazione in Data Lineage, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale. 

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Go nella guida rapida di Data Lineage per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Data Lineage Go.

Per eseguire l'autenticazione in Data Lineage, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale. 


//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

Prima di provare questo esempio, segui le istruzioni di configurazione di Java nella guida rapida di Data Lineage per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Data Lineage Java.

Per eseguire l'autenticazione in Data Lineage, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale. 

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python nella guida rapida di Data Lineage per l'utilizzo delle librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Data Lineage Python.

Per eseguire l'autenticazione in Data Lineage, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale. 

# 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

Per visualizzare la configurazione della derivazione corrente, utilizza il comando gcloud datalineage config describe. Puoi recuperare la configurazione per un progetto, una cartella o un'organizzazione.

Il seguente esempio mostra come ottenere la configurazione per il progetto corrente:

gcloud datalineage config describe

Ad esempio, per ottenere la configurazione per un progetto specifico, utilizza il flag --project:

gcloud datalineage config describe --project=PROJECT_ID

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto di cui vuoi visualizzare la configurazione.

Per visualizzare la configurazione attuale dell'importazione della derivazione di un servizio per una cartella o un'organizzazione, sostituisci --project=PROJECT_ID con:

  • --folder=FOLDER_ID se vuoi visualizzare le impostazioni di importazione dati per una cartella.
  • --organization=ORGANIZATION_ID se vuoi visualizzare le impostazioni di importazione dati per un'organizzazione.

REST

Per visualizzare la configurazione della derivazione attuale, utilizza il metodo projects.locations.config.get. Puoi recuperare la configurazione per un progetto, una cartella o un'organizzazione.

L'esempio seguente mostra come ottenere la configurazione di un progetto:

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • CLIENT_PROJECT_ID: L'ID del progetto client utilizzato per la fatturazione o le quote.
  • PROJECT_ID: l'ID del progetto di cui vuoi visualizzare la configurazione.

Metodo HTTP e URL: 

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

Per inviare la richiesta, espandi una di queste opzioni:

Il comando restituisce uno dei seguenti output:

  • Se non fornisci impostazioni di importazione della derivazione, riceverai un output con un oggetto ingestion vuoto: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {}
}

    Ciò significa che il servizio utilizza l'impostazione predefinita di importazione della derivazione. In questo esempio, l'impostazione di importazione della derivazione per Managed Service for Apache Spark è enabled.

  • Se attivi l'importazione della derivazione in modo esplicito, ottieni il seguente output: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": true
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}
  • Se l'importazione della derivazione è disattivata, viene visualizzato il seguente output: 
    {
  "name": "projects/123456789012/locations/global/config",
  "ingestion": {
    "rules": [
      {
        "integrationSelector": {
          "integration": "DATAPROC"
        },
        "lineageEnablement": {
          "enabled": false
        }
      }
    ]
  },
  "etag": "1a2b3c4d5e"
}

Per ottenere la configurazione di una cartella o di un'organizzazione, sostituisci projects/PROJECT_ID con folders/FOLDER_ID o organizations/ORGANIZATION_ID.

Il campo etag nella risposta è un checksum generato dal server in base al valore attuale della configurazione. Quando aggiorni una configurazione utilizzando il metodo patch, puoi includere il valore etag restituito da una recente richiesta get nel corpo della richiesta. Se fornisci etag, Knowledge Catalog lo utilizza per verificare che la configurazione non sia cambiata dall'ultima richiesta di lettura. Se non corrispondono, la richiesta di aggiornamento non va a buon fine. In questo modo eviti di sovrascrivere involontariamente le configurazioni apportate da altri utenti negli scenari di lettura-modifica-scrittura. Se non fornisci un etag nella tua richiesta patch, Knowledge Catalog sovrascrive la configurazione in modo incondizionato.

Disabilita l'importazione della derivazione per un servizio

Per gestire i costi, applicare le norme di governance dei dati o escludere i progetti di sviluppo e altri carichi di lavoro che non traggono vantaggio dal monitoraggio della derivazione, disattiva l'importazione della derivazione per un servizio.

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

Per disattivare l'importazione della derivazione per un servizio specifico, utilizza il comando gcloud datalineage config update con una stringa JSON incorporata o un percorso a un file JSON che imposta lineageEnablement.enabled su false per integration specifico.

L'esempio seguente mostra come disattivare l'importazione della derivazione di un servizio per un progetto utilizzando una stringa JSON incorporata:

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

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto di cui vuoi aggiornare la configurazione.
  • INTEGRATION: l'integrazione per cui hai impostato la configurazione. Ad esempio: DATAPROC o BIGQUERY.
  • ETAG: il valore etag restituito da una recente richiesta get nel corpo della richiesta, utilizzato per verificare che la configurazione non sia cambiata dall'ultima richiesta di lettura.

Per aggiornare la configurazione utilizzando un file JSON, esegui:

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

Sostituisci quanto segue:

  • CONFIG_FILE: il percorso del file JSON contenente la configurazione.

Per disattivare l'importazione della derivazione di un servizio per una cartella o un'organizzazione, sostituisci --project=PROJECT_ID con:

  • --folder=FOLDER_ID se vuoi aggiornare le impostazioni di importazione dati per una cartella.
  • --organization=ORGANIZATION_ID se vuoi aggiornare le impostazioni di importazione dati per un'organizzazione.

REST

Per disattivare l'importazione della derivazione per un servizio specifico, utilizza il metodo projects.locations.config.patch con una regola di importazione che imposta lineageEnablement.enabled su false per integration specifico.

Per evitare di sovrascrivere involontariamente le configurazioni apportate da altri utenti negli scenari di lettura-modifica-scrittura, puoi includere il campo etag nel corpo della richiesta. Per saperne di più, consulta la sezione Recuperare la configurazione attuale.

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • CLIENT_PROJECT_ID: L'ID del progetto client utilizzato per la fatturazione o le quote.
  • PROJECT_ID: l'ID del progetto di cui vuoi aggiornare la configurazione.
  • ETAG: il valore etag restituito da una recente richiesta get.
  • INTEGRATION: il integration per cui hai impostato la configurazione. Ad esempio: DATAPROC.

Metodo HTTP e URL: 

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

Corpo JSON della richiesta: 

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

Per inviare la richiesta, espandi una di queste opzioni:

Dovresti ricevere una risposta JSON simile alla seguente:

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

Per disattivare l'importazione della derivazione per una cartella o un'organizzazione, sostituisci projects/PROJECT_ID con folders/FOLDER_ID o organizations/ORGANIZATION_ID.

Abilita l'importazione della derivazione per un servizio

Per riprendere il monitoraggio dopo averlo disattivato o per attivare un'integrazione disattivata per impostazione predefinita, attiva l'importazione della derivazione per un servizio.

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

Per attivare l'importazione della derivazione per un servizio specifico, utilizza il comando gcloud datalineage config update con una stringa JSON incorporata o un percorso a un file JSON che imposta lineageEnablement.enabled su true per integration specifico. Le integrazioni attuali includono Managed Service for Apache Spark, BigQuery e Managed Airflow.

L'esempio seguente mostra come attivare l'importazione della derivazione di un servizio per un progetto utilizzando una stringa JSON incorporata:

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

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto di cui vuoi aggiornare la configurazione.
  • INTEGRATION: l'integrazione per cui hai impostato la configurazione (ad esempio DATAPROC o BIGQUERY).
  • ETAG: il valore etag restituito da una recente richiesta get nel corpo della richiesta, utilizzato per verificare che la configurazione non sia cambiata dall'ultima richiesta di lettura.

Per aggiornare la configurazione utilizzando un file JSON, esegui:

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

Sostituisci quanto segue:

  • CONFIG_FILE: il percorso del file JSON contenente la configurazione.

Per attivare l'importazione della derivazione di un servizio per una cartella o un'organizzazione, sostituisci --project=PROJECT_ID con:

  • --folder=FOLDER_ID se vuoi aggiornare le impostazioni di importazione dati per una cartella.
  • --organization=ORGANIZATION_ID se vuoi aggiornare le impostazioni di importazione dati per un'organizzazione.

REST

Per attivare l'importazione della derivazione per un servizio specifico, utilizza il metodo projects.locations.config.patch con una regola di importazione che imposta lineageEnablement.enabled su true per lo specifico integration. Le integrazioni attuali includono Managed Service for Apache Spark, BigQuery e Managed Airflow.

Per evitare di sovrascrivere involontariamente le configurazioni apportate da altri utenti negli scenari di lettura-modifica-scrittura, puoi includere il campo etag nel corpo della richiesta. Per saperne di più, consulta la sezione Recuperare la configurazione attuale.

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • CLIENT_PROJECT_ID: L'ID del progetto client utilizzato per la fatturazione o le quote.
  • PROJECT_ID: l'ID del progetto di cui vuoi aggiornare la configurazione.
  • ETAG: il valore etag restituito da una recente richiesta get.
  • INTEGRATION: il integration per cui hai impostato la configurazione. Ad esempio: DATAPROC.

Metodo HTTP e URL: 

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

Corpo JSON della richiesta: 

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

Per inviare la richiesta, espandi una di queste opzioni:

Dovresti ricevere una risposta JSON simile alla seguente:

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

Per attivare l'importazione della derivazione di un servizio per una cartella o un'organizzazione, sostituisci projects/PROJECT_ID con folders/FOLDER_ID o organizations/ORGANIZATION_ID.

Visualizza derivazione

Per monitorare la trasformazione e lo spostamento dei dati nei sistemi, puoi visualizzare la relativa derivazione utilizzando la console Google Cloud o l'API.

Console

Puoi accedere alle informazioni sulla derivazione dei dati nella Google Cloud console da vari punti di partenza:

  • Knowledge Catalog:vai alla pagina Ricerca di Knowledge Catalog, seleziona Knowledge Catalog come modalità di ricerca, cerca la voce che vuoi visualizzare e poi fai clic. Per saperne di più, consulta Cercare risorse in Knowledge Catalog.
  • BigQuery:vai alla pagina BigQuery e apri la tabella per cui vuoi visualizzare la derivazione dei dati.
  • Vertex AI: vai alla pagina Set di dati o Registro dei modelli e fai clic sul set di dati o sul modello per cui vuoi visualizzare la derivazione dei dati.

Per visualizzare il grafico della derivazione:

  1. Fai clic sulla scheda Lignaggio.

    Si apre la visualizzazione predefinita Grafico, che mostra la derivazione a livello di tabella in tutti i sistemi e le regioni. Per saperne di più, consulta la sezione Visualizzazione del grafico della derivazione.

  2. Per esplorare manualmente il grafico della derivazione, fai clic su Espandi accanto a un nodo per caricare altri cinque nodi alla volta.

    Per saperne di più, consulta Esplorare manualmente il grafico della derivazione.

  3. Fai clic su un nodo nella visualizzazione Grafico.

    Si apre il riquadro Dettagli con informazioni sull'asset, ad esempio nome e tipo completamente qualificati. Per saperne di più, consulta Dettagli del nodo.

  4. Fai clic su un bordo con un'icona di processo nella visualizzazione Grafico.

    Si apre il riquadro Query. Per maggiori informazioni, vedi Esaminare la logica di trasformazione e Controllo e cronologia delle esecuzioni.

    • Per esaminare la logica di trasformazione, fai clic sulla scheda Dettagli.
    • Per visualizzare l'audit e la cronologia delle esecuzioni, fai clic sulla scheda Esecuzioni.

  5. Nel riquadro Esplora lignaggio, seleziona i criteri di filtro, ad esempio Direzione, Tipo di dipendenza o Intervallo di tempo, poi fai clic su Applica.

    Si apre una visualizzazione mirata all'interno di una regione specifica (anteprima). Questa visualizzazione espande automaticamente il grafico fino a tre livelli di nodi. Per saperne di più, consulta Applicare filtri per una visualizzazione della lineage mirata.

  6. Nella visualizzazione Grafico selezionata, seleziona un nodo e poi, nel riquadro dei dettagli del nodo, fai clic su Visualizza percorso per visualizzare il percorso di derivazione dal nodo selezionato all'elemento principale (solo nella visualizzazione selezionata).

    Per ulteriori informazioni, vedi Visualizzazione del percorso di derivazione.

  7. Per visualizzare la derivazione a livello di colonna (solo per i job BigQuery e Managed Service for Apache Spark), esegui una delle seguenti operazioni:

    • In una visualizzazione Grafico mirata, fai clic sull'icona della colonna in una tabella.
      Icona utilizzata per passare alla derivazione a livello di colonna.
      Icona colonna
    • Nel riquadro Esplora derivazioni, filtra per nome della colonna e fai clic su Applica.

    Per saperne di più, consulta la sezione Derivazione a livello di colonna.

  8. Fai clic su Reimposta.

    Questa azione rimuove tutti i filtri applicati e ti porta all'inizio della visualizzazione del grafico.

  9. Fai clic su Elenco per passare alla visualizzazione elenco.

    La visualizzazione Elenco offre rappresentazioni tabulari semplificate e dettagliate della derivazione sia a livello di tabella che di colonna, sincronizzate con la visualizzazione Grafico. Per impostazione predefinita, viene visualizzata la visualizzazione elenco semplificata e puoi passare alla visualizzazione elenco dettagliata per analizzare le singole relazioni origine-destinazione. Puoi configurare le colonne visualizzate ed esportare i dati di derivazione. Per saperne di più, consulta Visualizzazione elenco della derivazione.

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

Perfezionare la visualizzazione della derivazione

Per perfezionare la visualizzazione della derivazione, puoi utilizzare le opzioni di evidenziazione e filtro in Esplora derivazioni:

  1. Per cercare progetti, set di dati o nomi di entità specifici, utilizza il riquadro Filtri.

    Dopo aver applicato i filtri, i nodi di derivazione che corrispondono ai criteri di filtro vengono considerati nodi corrispondenti. Puoi perfezionare la modalità di visualizzazione dei nodi corrispondenti e non corrispondenti.

  2. Nel grafico della derivazione, fai clic sull'icona Altre azioni accanto al pulsante Cancella filtri per visualizzare le opzioni di visualizzazione.

  3. Seleziona una o entrambe le seguenti opzioni:

Opzioni di evidenziazione e filtro in Explorer di lineage.
Opzioni di evidenziazione e filtro.

Puoi selezionare entrambe le opzioni contemporaneamente. Se sono selezionate entrambe le opzioni, i nodi non filtrati vengono nascosti e i nodi corrispondenti vengono evidenziati nella visualizzazione del grafico filtrato.

Passaggi successivi