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. Cuando actúa como consumidor de OpenLineage con la API de REST de ProcessOpenLineageRunEvent, Knowledge Catalog te permite unificar el linaje de canalizaciones personalizadas junto con el linaje integrado de los servicios de Google Cloud .

Descripción general

OpenLineage es una plataforma abierta para recopilar y analizar información del 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 mostrarlos en la interfaz web de Knowledge Catalog junto con la información de linaje de los servicios deGoogle Cloud , como BigQuery, Managed Service para Apache Airflow, Cloud Data Fusion y Managed Service para Apache Spark.

Para importar eventos de OpenLineage que usen la especificación de OpenLineage, usa el método de la API de REST ProcessOpenLineageRunEvent 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 las versiones principales 1 y 2 de OpenLineage.

  • Acciones de la API: El extremo de API de Data Lineage ProcessOpenLineageRunEvent solo actúa como consumidor de mensajes de OpenLineage, no como productor. La API te permite enviar información de linaje generada por cualquier herramienta o sistema compatible con OpenLineage a Knowledge Catalog. Algunos servicios de Google Cloud , 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 admitidas: La API de Data Lineage no admite lo siguiente:

    • Cualquier versión posterior de OpenLineage con cambios en el formato de los mensajes
    • 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.

  • Linaje a nivel de columna: El extremo ProcessOpenLineageRunEvent materializa los bordes de linaje a nivel de tabla solo para los eventos personalizados de OpenLineage. Los gráficos de linaje a nivel de columna solo se generan para las transformaciones de SQL de BigQuery y los trabajos de Managed Service para Apache Spark. Las facetas personalizadas a nivel de columna no se renderizan en la consola.

  • 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. En el caso de los motores automatizados compatibles que recopilan linaje a nivel de columna, 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, en el que se indican 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 Cómo comenzar.

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 para C# que encontrarás en la guía de inicio rápido sobre el linaje de datos con bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data Lineage para C#.

Para autenticarte en el linaje de datos, 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 para Go que encontrarás en la guía de inicio rápido sobre el linaje de datos con bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data Lineage para Go.

Para autenticarte en el linaje de datos, 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 para Java que encontrarás en la guía de inicio rápido sobre el linaje de datos con bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data Lineage para Java.

Para autenticarte en el linaje de datos, 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 para Python que encontrarás en la guía de inicio rápido sobre el linaje de datos con bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data Lineage para Python.

Para autenticarte en el linaje de datos, 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 para Ruby que encontrarás en la guía de inicio rápido sobre el linaje de datos con bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Data Lineage para Ruby.

Para autenticarte en el linaje de datos, 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 método processOpenLineageRunEvent.

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

  • PROJECT_ID: Es el ID del proyecto de Google Cloud .
  • LOCATION_ID: Es la ubicación de Google Cloud , 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 el linaje de datos: 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 productores 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 Java del productor para el linaje de datos ahora es de código abierto. La biblioteca está disponible en GitHub y Maven.
  • Transporte de OpenLineage para GCP: Para los productores de OpenLineage basados en Java, hay disponible un transporte de GcpLineage dedicado. Simplifica la integración en 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 existente de OpenLineage, 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 Visualiza 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 faceta:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • todos los spark.logicalPlan.*
  • environment-properties (faceta de linaje Google Cloud personalizada)
    • 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?