Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Intégrer à OpenLineage

Ce document explique comment intégrer OpenLineage à Knowledge Catalog (anciennement Dataplex Universal Catalog) pour importer et visualiser la traçabilité des données à partir de systèmes externes. En agissant en tant que consommateur OpenLineage à l'aide de l'API REST ProcessOpenLineageRunEvent, Knowledge Catalog vous permet d'unifier la traçabilité des pipelines personnalisés avec la traçabilité intégrée des Google Cloud services.

Présentation

OpenLineage est une plate-forme ouverte permettant de collecter et d' analyser des informations sur la traçabilité des données. En utilisant une norme ouverte pour les données de traçabilité, OpenLineage capture les événements de traçabilité à partir des composants du pipeline de données qui utilisent une API OpenLineage pour générer des rapports sur les exécutions, les jobs et les ensembles de données.

Grâce à l'API Data Lineage, vous pouvez importer des événements OpenLineage à afficher dans l'interface Web de Knowledge Catalog, ainsi que des informations de traçabilité provenant de Google Cloud services tels que BigQuery, Managed Service pour Apache Airflow, Cloud Data Fusion et Managed Service pour Apache Spark.

Pour importer des événements OpenLineage qui utilisent la spécification OpenLineage, utilisez la ProcessOpenLineageRunEvent méthode d'API REST et mappez les facettes OpenLineage aux attributs de l'API Data Lineage.

Limites d'intégration d'OpenLineage

Versions compatibles : l'API Data Lineage est compatible avec la version majeure 1 d'OpenLineage.
Actions de l'API : le point de terminaison de l'API Data Lineage ProcessOpenLineageRunEvent agit uniquement en tant que consommateur de messages OpenLineage, et non en tant que producteur. L'API vous permet d'envoyer des informations de traçabilité générées par n'importe quel outil ou système compatible avec OpenLineage dans Knowledge Catalog. Certains Google Cloud services, tels que Managed Service pour Apache Spark et Managed Airflow, incluent des producteurs OpenLineage intégrés qui peuvent envoyer des événements à ce point de terminaison, ce qui automatise la capture de la traçabilité à partir de ces services.
Fonctionnalités non compatibles : l'API Data Lineage n'est pas compatible avec les éléments suivants :
- Toute version ultérieure d'OpenLineage avec des modifications du format de message
- DatasetEvent
- JobEvent
Taille des messages : la taille maximale d'un message est de 5 Mo.
Longueur des noms : la longueur de chaque nom complet dans les entrées et les sorties est limitée à 4 000 caractères.
Limites de liens Les liens sont regroupés par événements, avec un maximum de 100 liens par événement. Le nombre total maximal de liens au niveau des tables est de 1 000. Si un message contient plus de 1 500 liens au niveau des colonnes, les informations au niveau des colonnes sont ignorées.
Champ d'application du graphique : Knowledge Catalog affiche un graphique de traçabilité pour chaque exécution de job, indiquant les entrées et les sorties des événements de traçabilité. Il n'est pas compatible avec les processus de niveau inférieur, tels que les étapes Spark.

Mappage des attributs de facette OpenLineage

Pour en savoir plus sur le mappage OpenLineage, consultez Mappage OpenLineage.

Importer un événement OpenLineage

Si vous n'avez pas encore configuré OpenLineage, consultez Premiers pas.

Pour importer un événement OpenLineage dans Knowledge Catalog, appelez la méthode d'API ProcessOpenLineageRunEvent.

C#

Avant d'essayer cet exemple, suivez les instructions de configuration pour C# décrites dans le guide de démarrage rapide de Data Lineage à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l' API Data Lineage C#.

Pour vous authentifier auprès de Data Lineage, configurez le service Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer l'authentification pour un environnement de développement 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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Go décrites dans le guide de démarrage rapide de Data Lineage à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l' API Data Lineage Go.


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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Java décrites dans le guide de démarrage rapide de Data Lineage à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l' API Java Data Lineage.

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

Avant d'essayer cet exemple, suivez les instructions de configuration Python décrites dans le guide de démarrage rapide de Data Lineage à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l' API PythonData Lineage.

# 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

Avant d'essayer cet exemple, suivez les instructions de configuration pour Ruby décrites dans le guide de démarrage rapide de Data Lineage à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l' API Ruby Data Lineage.

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

Pour importer un événement OpenLineage, utilisez la processOpenLineageRunEvent méthode.

Avant d'utiliser les données de requête ci-dessous, effectuez les remplacements suivants :

PROJECT_ID: ID de votre Google Cloud projet.
LOCATION_ID: l' Google Cloud emplacement, par exemple us-central1.

Méthode HTTP et URL :

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

Corps JSON de la requête :

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

Pour envoyer votre requête, développez l'une des options suivantes :

curl (Linux, macOS ou Cloud Shell)

Remarque : La commande suivante suppose que vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login, ou en utilisant Cloud Shell, qui vous connecte automatiquement à la CLI gcloud. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Enregistrez le corps de la requête dans un fichier nommé request.json, puis exécutez la commande suivante :

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent"

PowerShell (Windows)

Remarque : La commande suivante suppose que vous vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login. Vous pouvez exécuter la commande gcloud auth list pour vérifier quel est le compte actuellement actif.

Enregistrez le corps de la requête dans un fichier nommé request.json, puis exécutez la commande suivante :

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent" | Select-Object -Expand Content

Vous devriez recevoir une réponse JSON de ce type :

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

Outils pour envoyer des messages OpenLineage

Pour simplifier l'envoi d'événements à l'API Data Lineage, vous pouvez utiliser différents outils et bibliothèques :

Bibliothèques clientes Google Cloud pour Data Lineage : Google fournit des bibliothèques clientes pour interagir avec l'API Data Lineage de manière automatisée. Pour obtenir des instructions d'installation, consultez Bibliothèques clientes.
Bibliothèque de producteurs Java Google Cloud : Google fournit une bibliothèque Java Open Source pour vous aider à créer et à envoyer des événements OpenLineage à l'API Data Lineage. Pour en savoir plus, consultez l'article de blog La bibliothèque Java de producteurs pour Data Lineage est désormais Open Source. La bibliothèque est disponible sur GitHub et Maven.
Transport OpenLineage GCP : pour les producteurs OpenLineage basés sur Java, un transport GcpLineage dédié est disponible. Il simplifie l'intégration à l'API Data Lineage en réduisant le code nécessaire à l'envoi d'événements à l'API Data Lineage. GcpLineageTransport peut être configuré en tant que récepteur d'événements pour n'importe quel producteur OpenLineage existant, tel qu'Airflow, Spark et Flink. Pour en savoir plus et obtenir des exemples, consultez GcpLineage.

Analyser les informations d'OpenLineage

Pour analyser les événements OpenLineage importés, consultez Afficher les graphiques de traçabilité dans l'interface utilisateur de Knowledge Catalog.

Données de facette OpenLineage stockées

L'API Data Lineage ne stocke pas toutes les données de facette des messages OpenLineage. L'API Data Lineage stocke les champs de facette suivants :

spark_version
- openlineage-spark-version
- spark-version
Tous les spark.logicalPlan.*
environment-properties (facette custom Google Cloud lineage)
- origin.sourcetype et 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 stocke les informations suivantes :

eventTime
run.runId
job.namespace
job.name

Étape suivante

En savoir plus sur la traçabilité des données avec Managed Service pour Apache Spark et la traçabilité des données Hive intégrations.
Essayez-le dans un atelier interactif : Capturer et explorer les mises à jour de données avec Data Lineage et OpenLineage

Intégrer à OpenLineage Restez organisé à l'aide des collections Enregistrez et classez les contenus selon vos préférences.

Présentation

Limites d'intégration d'OpenLineage

Mappage des attributs de facette OpenLineage

Importer un événement OpenLineage

C#

C#

Go

Go

Java

Java

Python

Python

Ruby

Ruby

REST

curl (Linux, macOS ou Cloud Shell)

PowerShell (Windows)

Outils pour envoyer des messages OpenLineage

Analyser les informations d'OpenLineage

Données de facette OpenLineage stockées

Étape suivante

Intégrer à OpenLineage