Utiliser la traçabilité des données avec Managed Service pour Apache Spark

Ce document explique comment activer la traçabilité des données sur Managed Service pour Apache Spark pour les charges de travail par lot et les sessions interactives au niveau du projet, de la charge de travail par lot ou de la session interactive.

Présentation

La traçabilité des données est une fonctionnalité de Knowledge Catalog qui vous permet de suivre la façon dont les données transitent par vos systèmes : leur origine, leur destination, et les transformations qui leur sont appliquées.

Les charges de travail et les sessions Managed Service pour Apache Spark capturent les événements de traçabilité et les publient dans l'API Data Lineage de Knowledge Catalog. Managed Service pour Apache Spark s'intègre à l'API Data Lineage via OpenLineage, à l'aide du plug-in OpenLineage Spark.

Vous pouvez accéder aux informations de traçabilité via Knowledge Catalog, à l'aide de graphiques de traçabilité et de l'API Data Lineage. Pour en savoir plus, consultez Afficher les graphiques de traçabilité dans Knowledge Catalog.

Disponibilité

La traçabilité des données, qui est compatible avec les sources de données BigQuery et Cloud Storage est disponible pour les charges de travail et les sessions exécutées avec les versions d'exécution Managed Service pour Apache Spark compatibles à l'exception des cas suivants :

  • La traçabilité des données n'est pas disponible pour les charges de travail ni les sessions SparkR ou Spark Streaming.

Avant de commencer

  1. Sur la page de sélection du projet de la Google Cloud console, sélectionnez le projet à utiliser pour vos charges de travail ou sessions Managed Service pour Apache Spark.

    Accéder au sélecteur de projet

  2. Activez l'API Data Lineage.

    Activer les API

    Modifications à venir concernant la traçabilité des données Spark Consultez les notes de version de Managed Service pour Apache Spark pour connaître l'annonce d'une modification qui rendra automatiquement la traçabilité des données Spark disponible pour vos projets, charges de travail par lot et sessions interactives lorsque vous activez l'API Data Lineage (voir Contrôler l'ingestion de la traçabilité pour un service) sans nécessiter de paramètres supplémentaires pour le projet, la charge de travail par lot ou la session interactive.

Rôles requis

Si votre charge de travail par lot utilise le compte de service Managed Service pour Apache Spark par défaut, elle dispose du rôle Managed Service for Apache Spark Worker, qui contient les autorisations requises par la traçabilité des données.

Toutefois, si votre charge de travail par lot utilise un compte de service personnalisé pour activer la traçabilité des données, vous devez accorder l'un des rôles listés dans le paragraphe suivant, qui contiennent les autorisations requises par la traçabilité des données, au compte de service personnalisé.

Pour obtenir les autorisations nécessaires pour utiliser la traçabilité des données avec Managed Service pour Apache Spark, demandez à votre administrateur de vous accorder les rôles IAM suivants sur le compte de service personnalisé de votre charge de travail par lot :

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

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

Activer la traçabilité des données Spark

Vous pouvez activer la traçabilité des données Spark pour votre projet, votre charge de travail par lot ou votre session interactive.

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

Une fois que vous avez activé la traçabilité des données Spark au niveau du projet, les tâches Spark suivantes qui s'exécutent dans une charge de travail par lot ou une session interactive auront la traçabilité des données Spark activée.

Pour activer la traçabilité des données Spark sur votre projet, définissez les métadonnées de projet personnalisées suivantes :

Clé Valeur
DATAPROC_LINEAGE_ENABLED true

Vous pouvez désactiver la traçabilité des données Spark pour un projet en définissant les métadonnées DATAPROC_LINEAGE_ENABLED sur false.

Activer la traçabilité des données sur une charge de travail par lot Spark

Pour activer la traçabilité des données sur une charge de travail par lot, définissez la propriété spark.dataproc.lineage.enabled sur true lorsque vous envoyez la charge de travail. Ce paramètre remplace tout paramètre de traçabilité des données Spark au niveau du projet : si la traçabilité des données Spark est désactivée au niveau du projet, mais activée pour la charge de travail par lot, le paramètre de la charge de travail par lot est prioritaire.

Vous pouvez désactiver la traçabilité des données Spark sur une charge de travail par lot Spark en définissant la propriété spark.dataproc.lineage.enabled sur false lorsque vous envoyez la charge de travail.

Cet exemple utilise gcloud CLI pour envoyer une charge de travail par lot lineage-example.py avec la traçabilité Spark activée.

gcloud dataproc batches submit pyspark lineage-example.py \
    --region=REGION \
    --deps-bucket=gs://BUCKET \
    --properties=spark.dataproc.lineage.enabled=true

Le code lineage-example.py suivant lit les données d'une table BigQuery publique, puis écrit la sortie dans une nouvelle table d'un ensemble de données BigQuery existant. Il utilise un bucket Cloud Storage pour le stockage temporaire.

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

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

source = 'bigquery-public-data:samples.shakespeare'
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_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .option('writeMethod', 'direct') \
  .save()

Remplacez les éléments suivants :

  • REGION : région dans laquelle exécuter la charge de travail
  • BUCKET : nom d'un bucket Cloud Storage existant dans lequel stocker les dépendances
  • PROJECT_ID, DATASET, et TABLE : ID du projet, nom d'un ensemble de données BigQuery existant et nom d'une nouvelle table à créer dans l'ensemble de données (la table ne doit pas exister)

Vous pouvez afficher le graphique de traçabilité dans l'interface utilisateur de Knowledge Catalog.

Graphique de traçabilité Spark

Activer la traçabilité des données sur une session interactive ou un modèle de session Spark

Pour activer la traçabilité des données sur une session interactive ou un modèle de session Spark, définissez la propriété spark.dataproc.lineage.enabled sur true lorsque vous créez la session ou le modèle de session. Ce paramètre remplace tout paramètre de traçabilité des données Spark au niveau du projet : si la traçabilité des données Spark est désactivée au niveau du projet, mais activée pour la session interactive, le paramètre de la session interactive est prioritaire.

Vous pouvez désactiver la traçabilité des données Spark sur une session interactive ou un modèle de session Spark en définissant la propriété spark.dataproc.lineage.enabled sur false lorsque vous créez la session interactive ou le modèle de session.

Le code de notebook PySpark suivant configure une session interactive Managed Service pour Apache Spark avec la traçabilité des données Spark activée. Il crée ensuite une session Spark Connect qui exécute une requête de décompte de mots sur un ensemble de données BigQuery Shakespeare public, puis écrit la sortie dans une nouvelle table d'un ensemble de données BigQuery existant (voir Créer une session Spark dans un notebook BigQuery Studio) .

# Configure the Dataproc Serverless interactive session
# to enable Spark data lineage.
from google.cloud.dataproc_v1 import Session

session = Session()
session.runtime_config.properties["spark.dataproc.lineage.enabled"] = "true"

# Create the Spark Connect session.
from google.cloud.dataproc_spark_connect import DataprocSparkSession

spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()

# Run a wordcount query on the public BigQuery Shakespeare dataset.
source = "bigquery-public-data:samples.shakespeare"
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')

# Output the results to a BigQuery destination table.
destination_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .save()

Remplacez les éléments suivants :

  • PROJECT_ID, DATASET, et TABLE : ID du projet, nom d'un ensemble de données BigQuery existant et nom d'une nouvelle table à créer dans l'ensemble de données (la table ne doit pas exister)

Vous pouvez afficher le graphique de traçabilité des données en cliquant sur le nom de la table de destination dans le volet de navigation de la page Explorateur de BigQuery, puis en sélectionnant l'onglet "Traçabilité" dans le volet des détails de la table.

Graphique de traçabilité Spark

Afficher la traçabilité dans Knowledge Catalog

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

Étape suivante