Integra con OpenLineage

Questo documento spiega come integrare OpenLineage con Knowledge Catalog (in precedenza Dataplex Universal Catalog) per importare e visualizzare la tracciabilità dei dati da sistemi esterni. Agendo come consumer OpenLineage utilizzando l'API REST ProcessOpenLineageRunEvent, Knowledge Catalog ti consente di unificare la tracciabilità della pipeline personalizzata insieme alla tracciabilità integrata dei Google Cloud servizi.

Panoramica

OpenLineage è una piattaforma aperta per la raccolta e l'analisi delle informazioni sulla tracciabilità dei dati. Utilizzando uno standard aperto per i dati di tracciabilità, OpenLineage acquisisce gli eventi di tracciabilità dai componenti della pipeline di dati che utilizzano un'API OpenLineage per segnalare esecuzioni, job e set di dati.

Tramite l'API Data Lineage, puoi importare gli eventi OpenLineage da visualizzare nell'interfaccia web di Knowledge Catalog insieme alle informazioni sulla derivazione dei Google Cloud servizi, come BigQuery, Managed Service for Apache Airflow, Cloud Data Fusion e Managed Service for Apache Spark.

Per importare gli eventi OpenLineage che utilizzano la specifica OpenLineage, utilizza il ProcessOpenLineageRunEvent metodo dell'API REST e mappa le sfaccettature OpenLineage agli attributi dell'API Data Lineage.

Limitazioni dell'integrazione di OpenLineage

  • Versioni supportate: l'API Data Lineage supporta la versione principale 1 di OpenLineage.

  • Azioni API: l'endpoint API Data Lineage ProcessOpenLineageRunEvent funge solo da consumer dei messaggi OpenLineage, non da producer. L'API ti consente di inviare le informazioni sulla tracciabilità generate da qualsiasi strumento o sistema conforme a OpenLineage a Knowledge Catalog. Alcuni Google Cloud servizi, come Managed Service for Apache Spark e Managed Airflow, includono producer OpenLineage integrati che possono inviare eventi a questo endpoint, automatizzando l'acquisizione della tracciabilità da questi servizi.

  • Funzionalità non supportate: l'API Data Lineage non supporta quanto segue:

    • Qualsiasi release OpenLineage successiva con modifiche al formato dei messaggi
    • DatasetEvent
    • JobEvent

  • Dimensioni dei messaggi: la dimensione massima di un singolo messaggio è di 5 MB.

  • Lunghezza dei nomi: La lunghezza di ogni nome completo negli input e negli output è limitata a 4000 caratteri.

  • Limiti dei link: I link sono raggruppati per eventi, con un massimo di 100 link per evento. Il numero massimo aggregato di link a livello di tabella è 1000. Se un messaggio contiene più di 1500 link a livello di colonna, le informazioni a livello di colonna vengono ignorate.

  • Ambito del grafico: Knowledge Catalog mostra un grafico di derivazione per ogni esecuzione del job, mostrando gli input e gli output degli eventi di derivazione. Non supporta i processi di livello inferiore come le fasi di Spark.

Mappatura degli attributi delle sfaccettature OpenLineage

Per informazioni sulla mappatura di OpenLineage, consulta Mappatura di OpenLineage.

Importare un evento OpenLineage

Se non hai ancora configurato OpenLineage, consulta la Guida rapida.

Per importare un evento OpenLineage in Knowledge Catalog, chiama il metodo dell'API ProcessOpenLineageRunEvent.

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'APIC# Data Lineage.

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.V1;
using Google.Protobuf.WellKnownTypes;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for ProcessOpenLineageRunEvent</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 ProcessOpenLineageRunEventRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        ProcessOpenLineageRunEventRequest request = new ProcessOpenLineageRunEventRequest
        {
            Parent = "",
            OpenLineage = new Struct(),
        };
        // Make the request
        ProcessOpenLineageRunEventResponse response = lineageClient.ProcessOpenLineageRunEvent(request);
    }
}

Vai

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'Go API Data Lineage.

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"

	lineage "cloud.google.com/go/datacatalog/lineage/apiv1"
	lineagepb "cloud.google.com/go/datacatalog/lineage/apiv1/lineagepb"
)

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 := lineage.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

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

Java

Java

Prima di provare questo esempio, segui le istruzioni di configurazione 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.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventRequest;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventResponse;
import com.google.protobuf.Struct;

public class SyncProcessOpenLineageRunEvent {

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

  public static void syncProcessOpenLineageRunEvent() 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 (LineageClient lineageClient = LineageClient.create()) {
      ProcessOpenLineageRunEventRequest request =
          ProcessOpenLineageRunEventRequest.newBuilder()
              .setParent("parent-995424086")
              .setOpenLineage(Struct.newBuilder().build())
              .setRequestId("requestId693933066")
              .build();
      ProcessOpenLineageRunEventResponse response =
          lineageClient.processOpenLineageRunEvent(request);
    }
  }
}

Python

Python

Prima di provare questo esempio, segui le istruzioni di configurazione 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_v1


def sample_process_open_lineage_run_event():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.ProcessOpenLineageRunEventRequest(
        parent="parent_value",
    )

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

    # Handle the response
    print(response)

Ruby

Ruby

Prima di provare questo esempio, segui le istruzioni di configurazione di Ruby 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 Ruby.

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. 

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the process_open_lineage_run_event call in the Lineage service
#
# 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/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#process_open_lineage_run_event.
#
def process_open_lineage_run_event
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::ProcessOpenLineageRunEventRequest.new

  # Call the process_open_lineage_run_event method.
  result = client.process_open_lineage_run_event request

  # The returned object is of type Google::Cloud::DataCatalog::Lineage::V1::ProcessOpenLineageRunEventResponse.
  p result
end

REST

Per importare un evento OpenLineage, utilizza il processOpenLineageRunEvent metodo.

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

  • PROJECT_ID: il tuo Google Cloud ID progetto.
  • LOCATION_ID: la Google Cloud località, ad esempio us-central1.

Metodo HTTP e URL: 

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent

Corpo JSON della richiesta: 

{
  "eventTime": "2023-04-04T13:21:16.098Z",
  "eventType": "COMPLETE",
  "inputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "job": {
    "name": "somename",
    "namespace": "customnamespace"
  },
  "outputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "producer": "someproducer",
  "run": {
    "runId": "somerunid"
  },
  "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/$defs/RunEvent"
}

Per inviare la richiesta, espandi una di queste opzioni:

Dovresti ricevere una risposta JSON simile alla seguente:

{
  "process": "projects/my-project/locations/us-central1/processes/my-process",
  "run": "projects/my-project/locations/us-central1/processes/my-process/runs/my-run",
  "lineageEvents": [
    "projects/my-project/locations/us-central1/processes/my-process/runs/my-run/lineageEvents/my-lineage-event"
  ]
}

Strumenti per l'invio di messaggi OpenLineage

Per semplificare l'invio di eventi all'API Data Lineage, puoi utilizzare vari strumenti e librerie:

  • Librerie client di Google Cloud per Data Lineage: Google fornisce librerie client per interagire con l'API Data Lineage a livello di programmazione. Per le istruzioni di installazione, consulta Librerie client.
  • Libreria Java Producer di Google Cloud: Google fornisce una libreria Java open source per aiutarti a creare e inviare eventi OpenLineage all'API Data Lineage. Per saperne di più, consulta il post del blog La libreria Java Producer per Data Lineage è ora open source. La libreria è disponibile su GitHub e Maven.
  • Trasporto OpenLineage GCP: per i producer OpenLineage basati su Java, è disponibile un trasporto GcpLineage dedicato. Semplifica l'integrazione con l'API Data Lineage, riducendo al minimo il codice necessario per l'invio di eventi all'API Data Lineage. GcpLineageTransport può essere configurato come sink di eventi per qualsiasi producer OpenLineage esistente, come Airflow, Spark e Flink. Per ulteriori informazioni ed esempi, consulta GcpLineage.

Analizzare le informazioni di OpenLineage

Per analizzare gli eventi OpenLineage importati, consulta Visualizzare i grafici di tracciabilità nell'interfaccia utente di Knowledge Catalog.

Dati delle sfaccettature OpenLineage archiviati

L'API Data Lineage non archivia tutti i dati delle sfaccettature dei messaggi OpenLineage. L'API Data Lineage archivia i seguenti campi delle sfaccettature:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • tutti spark.logicalPlan.*
  • environment-properties (custom Google Cloud lineage sfaccettatura)
    • origin.sourcetype e origin.name
    • spark.app.id
    • spark.app.name
    • spark.batch.id
    • spark.batch.uuid
    • spark.cluster.name
    • spark.cluster.region
    • spark.job.id
    • spark.job.uuid
    • spark.project.id
    • spark.query.node.name
    • spark.session.id
    • spark.session.uuid

L'API Data Lineage archivia le seguenti informazioni:

  • eventTime
  • run.runId
  • job.namespace
  • job.name

Passaggi successivi