Activer la traçabilité des données dans Dataproc

Ce document explique comment activer la lignée de données pour vos tâches Spark Dataproc au niveau du projet ou du cluster.

Présentation

La traçabilité des données est une fonctionnalité Dataplex qui vous permet de suivre la manière dont les données transitent par vos systèmes: leur origine, la cible de transmission, et les transformations qui leur sont appliquées.

La lignée de données est disponible pour toutes les tâches Spark Dataproc, à l'exception de SparkR, avec les images Dataproc Compute Engine 2.0.74 et versions ultérieures, et est compatible avec les sources de données BigQuery et Cloud Storage.

Une fois la fonctionnalité activée dans votre cluster Dataproc, les tâches Spark Dataproc capturent les événements de la lignée de données et les publient dans l'API Data Lineage Dataplex. Dataproc s'intègre à l'API Data Lineage via OpenLineage, à l'aide du plug-in OpenLineage Spark.

Vous pouvez accéder aux informations sur la traçabilité des données via Dataplex, à l'aide des éléments suivants:

Limites

La traçabilité des données n'est pas disponible pour les jobs SparkR ou Spark en streaming.

Avant de commencer

  1. Dans la console Google Cloud, sur la page de sélection du projet, sélectionnez le projet contenant le cluster Dataproc pour lequel vous souhaitez suivre le lignage.

    Accéder au sélecteur de projet

  2. Activez les API Data Lineage et Data Catalog.

    Activer les API

Rôles requis

Pour obtenir les autorisations nécessaires pour utiliser la lignée de données dans Dataproc, demandez à votre administrateur de vous accorder les rôles IAM suivants sur le compte de service de VM du cluster Dataproc:

Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

Vous pouvez également obtenir les autorisations requises via des rôles personnalisés ou d'autres rôles prédéfinis.

Activer la traçabilité des données au niveau du projet

Vous pouvez activer la traçabilité des données au niveau du projet. La traçabilité des données est activée pour les jobs Spark compatibles exécutés sur des clusters créés après l'activation de la traçabilité des données dans un projet. Notez que la généalogie des données n'est pas activée pour les tâches exécutées sur des clusters existants (clusters créés avant l'activation de la généalogie des données au niveau du projet).

Activer la traçabilité des données au niveau du projet

Pour activer la généalogie des données au niveau du projet, définissez les métadonnées de projet personnalisées suivantes:

Clé Valeur
DATAPROC_LINEAGE_ENABLED true
DATAPROC_CLUSTER_SCOPES https://www.googleapis.com/auth/cloud-platform

Vous pouvez désactiver la lignée des données au niveau du projet en définissant les métadonnées DATAPROC_LINEAGE_ENABLED sur false.

Activer la traçabilité des données au niveau du cluster

Vous pouvez activer la lignée de données lorsque vous créez un cluster afin que la lignée de données soit activée pour toutes les tâches Spark compatibles envoyées au cluster.

Activer la traçabilité des données au niveau du cluster

Pour activer la lignée de données sur un cluster, créez un cluster Dataproc avec la propriété de cluster dataproc:dataproc.lineage.enabled définie sur true.

Clusters avec version d'image 2.0:l'étendue cloud-platform d'accès aux VM du cluster Dataproc est requise pour la lignée des données. Les clusters de version d'image Dataproc créés avec la version d'image 2.1 ou ultérieure ont cloud-platform activé. Si vous spécifiez la version d'image Dataproc 2.0 lorsque vous créez un cluster, définissez le champ d'application sur cloud-platform.

Exemple de gcloud CLI:

gcloud dataproc clusters create CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --properties 'dataproc:dataproc.lineage.enabled=true'

Désactiver la traçabilité des données d'une tâche

Si vous activez la lignée de données au niveau du cluster, vous pouvez la désactiver pour une tâche spécifique en transmettant la propriété spark.extraListeners avec une valeur vide ("") lorsque vous envoyez la tâche.

Une fois activée, vous ne pouvez plus désactiver la lignée des données sur le cluster. Pour éliminer la lignée des données sur toutes les tâches de cluster, vous pouvez recréer le cluster sans la propriété dataproc:dataproc.lineage.enabled.

Envoyer une tâche Spark

Lorsque vous envoyez une tâche Spark sur un cluster Dataproc créé avec la filiation des données activée, Dataproc capture et signale les informations sur la filiation des données à l'API Data Lineage.

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --class CLASS \
    --jars=gs://APPLICATION_BUCKET/spark-application.jar \
    --properties=spark.openlineage.namespace=CUSTOM_NAMESPACE,spark.openlineage.appName=CUSTOM_APPNAME

Remarques :

  • L'ajout des propriétés spark.openlineage.namespace et spark.openlineage.appName, qui permettent d'identifier de manière unique la tâche, est facultatif. Si vous n'ajoutez pas ces propriétés, Dataproc utilise les valeurs par défaut suivantes :
    • Valeur par défaut pour spark.openlineage.namespace: PROJECT_ID
    • Valeur par défaut pour spark.openlineage.appName: spark.app.name

Afficher les graphiques de traçabilité dans Dataplex

Un graphique de visualisation de la lignée affiche les relations entre les ressources de votre projet et les processus qui les ont créées. Vous pouvez afficher les informations sur la traçabilité des données sous la forme d'une visualisation graphique dans la console Google Cloud ou les récupérer à partir de l'API Data Lineage sous la forme de données JSON.

Pour en savoir plus, consultez Afficher des graphiques de traçabilité dans l'interface utilisateur de Dataplex.

Exemple :

La tâche Spark suivante lit les données d'une table BigQuery et les écrit dans une autre table BigQuery.

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

spark = SparkSession \
  .builder \
  .appName('LINEAGE_BQ_TO_BQ') \
  .getOrCreate()

bucket = lineage-ol-test
spark.conf.set('temporaryGcsBucket', bucket)

source = sample.source
words = spark.read.format('bigquery') \
  .option('table', source) \
  .load()
words.createOrReplaceTempView('words')

word_count = spark.sql('SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')

destination = sample.destination
word_count.write.format('bigquery') \
  .option('table', destination) \
  .save()

Le job Spark crée le graphique de lignée suivant dans l'UI Dataplex:

Exemple de graphique de traçabilité

Étape suivante