Integración con OpenLineage

En este documento, se explica cómo integrar OpenLineage con Knowledge Catalog (antes Dataplex Universal Catalog) para importar y visualizar el linaje de datos de sistemas externos. Al actuar como consumidor de OpenLineage con la API de REST ProcessOpenLineageRunEvent, Knowledge Catalog te permite unificar el linaje de canalización personalizado junto con el linaje integrado de los Google Cloud servicios.

Descripción general

OpenLineage es una plataforma abierta para recopilar y analizar información de linaje de datos. Con un estándar abierto para los datos de linaje, OpenLineage captura eventos de linaje de los componentes de la canalización de datos que usan una API de OpenLineage para informar sobre ejecuciones, trabajos y conjuntos de datos.

A través de la API de Data Lineage, puedes importar eventos de OpenLineage para que se muestren en la interfaz web de Knowledge Catalog junto con la información de linaje de Google Cloud servicios como BigQuery, Managed Service para Apache Airflow, Cloud Data Fusion y Managed Service para Apache Spark.

Para importar eventos de OpenLineage que usan la especificación de OpenLineage, usa el método ProcessOpenLineageRunEvent de la API de REST y asigna las facetas de OpenLineage a los atributos de la API de Data Lineage.

Limitaciones de la integración de OpenLineage

  • Versiones compatibles: La API de Data Lineage admite la versión principal 1 de OpenLineage.

  • Acciones de la API: El extremo de API de Data Lineage ProcessOpenLineageRunEvent solo actúa como un consumidor de mensajes de OpenLineage, no como un productor. La API te permite enviar información de linaje generada por cualquier herramienta o sistema compatible con OpenLineage a Knowledge Catalog. Algunos Google Cloud servicios, como Managed Service para Apache Spark y Managed Airflow, incluyen **productores**de OpenLineage integrados que pueden enviar eventos a este extremo, lo que automatiza la captura de linaje de esos servicios.

  • Funciones no compatibles: La API de Data Lineage no admite lo siguiente:

    • Cualquier versión posterior de OpenLineage con cambios en el formato del mensaje
    • DatasetEvent
    • JobEvent

  • Tamaño del mensaje: El tamaño máximo de un solo mensaje es de 5 MB.

  • Longitud del nombre: La longitud de cada nombre completamente calificado en las entradas y salidas se limita a 4,000 caracteres.

  • Límites de vínculos: Los vínculos se agrupan por eventos, con un máximo de 100 vínculos por evento. La cantidad máxima agregada de vínculos a nivel de tabla es de 1,000. Si un mensaje contiene más de 1,500 vínculos a nivel de columna, se omite la información a nivel de columna.

  • Alcance del gráfico: Knowledge Catalog muestra un gráfico de linaje para cada ejecución de trabajo, que muestra las entradas y salidas de los eventos de linaje. No admite procesos de nivel inferior, como las etapas de Spark.

Asignación de atributos de facetas de OpenLineage

Para obtener información sobre la asignación de OpenLineage, consulta Asignación de OpenLineage.

Importa un evento de OpenLineage

Si aún no configuraste OpenLineage, consulta Comienza.

Para importar un evento de OpenLineage a Knowledge Catalog, llama al método de la API ProcessOpenLineageRunEvent.

C#

C#

Antes de probar este ejemplo, sigue las instrucciones de configuración de C# que encontrarás en la guía de inicio rápido de Data Lineage sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de C#Data Lineage.

Para autenticarte en Data Lineage, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

Go

Go

Antes de probar este ejemplo, sigue las instrucciones de configuración de Go que encontrarás en la guía de inicio rápido de Data Lineage sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data LineageGo.

Para autenticarte en Data Lineage, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 


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

Antes de probar este ejemplo, sigue las instrucciones de configuración de Java que encontrarás en la guía de inicio rápido de Data Lineage sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de JavaData Lineage.

Para autenticarte en Data Lineage, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

Antes de probar este ejemplo, sigue las instrucciones de configuración de Python que encontrarás en la guía de inicio rápido de Data Lineage sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de PythonData Lineage.

Para autenticarte en Data Lineage, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

# 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

Antes de probar este ejemplo, sigue las instrucciones de configuración de Ruby que encontrarás en la guía de inicio rápido de Data Lineage sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de RubyData Lineage.

Para autenticarte en Data Lineage, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

Para importar un evento de OpenLineage, usa el processOpenLineageRunEvent método.

Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:

  • PROJECT_ID: Es el ID del Google Cloud proyecto de.
  • LOCATION_ID: Es la Google Cloud ubicación, como us-central1.

Método HTTP y URL: 

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

Cuerpo JSON de la solicitud: 

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

Para enviar tu solicitud, expande una de estas opciones:

Deberías recibir una respuesta JSON similar a la que se muestra a continuación:

{
  "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"
  ]
}

Herramientas para enviar mensajes de OpenLineage

Para simplificar el envío de eventos a la API de Data Lineage, puedes usar varias herramientas y bibliotecas:

  • Bibliotecas cliente de Google Cloud para Data Lineage: Google proporciona bibliotecas cliente para interactuar con la API de Data Lineage de manera programática. Para obtener instrucciones de instalación, consulta Bibliotecas cliente.
  • Biblioteca de productor de Java de Google Cloud: Google proporciona una biblioteca de Java de código abierto para ayudar a construir y enviar eventos de OpenLineage a la API de Data Lineage. Para obtener más información, consulta la entrada de blog La biblioteca de productor de Java para Data Lineage ahora es de código abierto. La biblioteca está disponible en GitHub y Maven.
  • Transporte de OpenLineage GCP: Para los productores de OpenLineage basados en Java, está disponible un transporte de GcpLineage dedicado. Simplifica la integración con la API de Data Lineage, ya que minimiza el código necesario para enviar eventos a la API de Data Lineage. El GcpLineageTransport se puede configurar como el receptor de eventos para cualquier productor de OpenLineage existente, como Airflow, Spark y Flink. Para obtener más información y ejemplos, consulta GcpLineage.

Analiza la información de OpenLineage

Para analizar los eventos de OpenLineage importados, consulta Cómo ver gráficos de linaje en la IU de Knowledge Catalog.

Datos de facetas de OpenLineage almacenados

La API de Data Lineage no almacena todos los datos de facetas de los mensajes de OpenLineage. La API de Data Lineage almacena los siguientes campos de facetas:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • todos los spark.logicalPlan.*
  • environment-properties (custom Google Cloud lineage faceta)
    • origin.sourcetype y 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

La API de Data Lineage almacena la siguiente información:

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

¿Qué sigue?