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 servizi Google Cloud .

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 generare report su esecuzioni, job e set di dati.

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

Per importare eventi OpenLineage che utilizzano la specifica OpenLineage, utilizza il metodo ProcessOpenLineageRunEvent 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 di messaggi OpenLineage, non da producer. L'API ti consente di inviare a Knowledge Catalog le informazioni sulla derivazione generate da qualsiasi strumento o sistema conforme a OpenLineage. Alcuni servizi Google Cloud , come Managed Service for Apache Spark e Managed Airflow, includono produttori 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 successiva di OpenLineage con modifiche al formato del messaggio
    • DatasetEvent
    • JobEvent

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

  • Lunghezza del nome:la lunghezza di ogni nome completo negli input e negli output è limitata a 4000 caratteri.

  • Tracciabilità a livello di colonna:l'endpoint ProcessOpenLineageRunEvent materializza i bordi della tracciabilità a livello di tabella solo per gli eventi OpenLineage personalizzati. I grafici di derivazione a livello di colonna vengono generati solo per le trasformazioni SQL di BigQuery e i job Managed Service for Apache Spark. I facet a livello di colonna personalizzati non vengono visualizzati nella console.

  • Limiti dei link:i link sono raggruppati per eventi, con un massimo di 100 link per evento. Il numero aggregato massimo di link a livello di tabella è 1000. Per i motori automatici supportati che raccolgono la derivazione a livello di colonna, 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 della tracciabilità per ogni esecuzione del job, mostrando gli input e gli output degli eventi di tracciabilità. Non supporta processi di livello inferiore come le fasi Spark.

Mappatura degli attributi dei facet OpenLineage

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

Importare un evento OpenLineage

Se non hai ancora configurato OpenLineage, consulta la sezione Inizia.

Per importare un evento OpenLineage in Knowledge Catalog, chiama il metodo 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'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.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'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"

	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 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.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 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_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 metodo processOpenLineageRunEvent.

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

  • PROJECT_ID: il tuo ID progetto Google Cloud .
  • LOCATION_ID: la posizione Google Cloud , 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 istruzioni di installazione, consulta Librerie client.
  • Libreria di produzione Java di Google Cloud:Google fornisce una libreria Java open source per creare e inviare eventi OpenLineage all'API Data Lineage. Per saperne di più, consulta il post del blog La libreria Java del produttore per Data Lineage è ora open source. La libreria è disponibile su GitHub e Maven.
  • OpenLineage GCP Transport: per i produttori OpenLineage basati su Java, è disponibile un GcpLineage Transport dedicato. Semplifica l'integrazione con l'API Data Lineage riducendo al minimo il codice necessario per inviare eventi all'API Data Lineage. GcpLineageTransport può essere configurato come sink di eventi per qualsiasi produttore 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, vedi Visualizzare i grafici di tracciabilità nella UI 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 i spark.logicalPlan.*
  • environment-properties (sfaccettatura della tracciabilità personalizzata Google Cloud )
    • 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