Integra con OpenLineage

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

Panoramica

OpenLineage è una piattaforma aperta per la raccolta e l'analisi delle informazioni sulla derivazione dei dati. Utilizzando uno standard aperto per i dati di derivazione, OpenLineage acquisisce gli eventi di derivazione 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 gli eventi OpenLineage da visualizzare nell'interfaccia web di Knowledge Catalog insieme alle informazioni sulla derivazione dei serviziGoogle Cloud , 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 metodo API REST ProcessOpenLineageRunEvent e mappa i facet OpenLineage agli attributi dell'API Data Lineage.

Limitazioni dell'integrazione di OpenLineage

  • Versioni supportate: l'API Data Lineage supporta le versioni principali 1 e 2 di OpenLineage.

  • Azioni API: l'endpoint dell'API Data Lineage ProcessOpenLineageRunEvent funge solo da consumer dei messaggi OpenLineage, non da producer. L'API consente di inviare informazioni sulla derivazione generate da qualsiasi strumento o sistema conforme a OpenLineage a Knowledge Catalog. Alcuni servizi Google Cloud , come Managed Service for Apache Spark e Managed Airflow, includono producer OpenLineage integrati che possono inviare eventi a questo endpoint, automatizzando l'acquisizione della derivazione 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 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.

  • Derivazione a livello di colonna: l'endpoint ProcessOpenLineageRunEvent materializza gli archi di derivazione 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 personalizzati a livello di colonna non vengono visualizzati nella console.

  • Limiti dei link: i link sono raggruppati per evento, con un massimo di 100 link per evento. Il numero massimo complessivo 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 visualizza un grafico di derivazione per ogni esecuzione del job, mostrando gli input e gli output degli eventi di derivazione. Non supporta processi di livello inferiore, come le fasi di Spark.

Mappatura degli attributi dei facet OpenLineage

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

Importa 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 località 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 la derivazione dei dati: Google fornisce librerie client per interagire con l'API Data Lineage in modo programmatico. Per le istruzioni di installazione, consulta Librerie client.
  • Libreria Google Cloud Java Producer: Google fornisce una libreria Java open source per aiutare a creare e inviare eventi OpenLineage all'API Data Lineage. Per saperne di più, consulta il blog post La libreria Java Producer per la derivazione dei dati è ora open source. La libreria è disponibile su GitHub e Maven.
  • OpenLineage GCP Transport: per i producer 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 producer OpenLineage esistente, come Airflow, Spark e Flink. Per ulteriori informazioni ed esempi, consulta GcpLineage.

Analizza le informazioni di OpenLineage

Per analizzare gli eventi OpenLineage importati, consulta Visualizza i grafici di derivazione nell'interfaccia utente di Knowledge Catalog.

Dati dei facet OpenLineage archiviati

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

  • spark_version
    • openlineage-spark-version
    • spark-version
  • tutti i spark.logicalPlan.*
  • environment-properties (facet di derivazione Google Cloud personalizzato)
    • 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