Utilizzare la derivazione dei dati con Managed Service per Apache Spark

Questo documento descrive come abilitare la derivazione dei dati sui workload batch e sulle sessioni interattive di Managed Service for Apache Spark a livello di progetto, workload batch o sessione interattiva o profilo di runtime (modello di sessione).

Panoramica

La derivazione dei dati è una funzionalità di Knowledge Catalog che consente di monitorare il modo in cui i dati vengono trasferiti nei sistemi: da dove provengono, dove vengono inviati e a quali trasformazioni sono sottoposti.

I workload e le sessioni di Managed Service for Apache Spark acquisiscono gli eventi di derivazione e li pubblicano nell'API Data Lineage di Knowledge Catalog. Managed Service for Apache Spark si integra con l'API Data Lineage tramite OpenLineage, utilizzando il plug-in OpenLineage Spark.

Puoi accedere alle informazioni sulla derivazione tramite Knowledge Catalog, utilizzando i grafici di derivazione e l'API Data Lineage. Per saperne di più, consulta Visualizza i grafici di derivazione in Knowledge Catalog.

Disponibilità

La derivazione dei dati, che supporta le origini dati BigQuery e Cloud Storage, è disponibile per i workload e le sessioni eseguiti con le versioni di runtime di Managed Service for Apache Spark supportate con le seguenti eccezioni e limitazioni:

  • La derivazione dei dati non è disponibile per le sessioni o i workload di flussi di dati SparkR o Spark.

Prima di iniziare

  1. Nella pagina di selezione del progetto nella console Google Cloud , seleziona il progetto da utilizzare per i workload o le sessioni di Managed Service for Apache Spark.

    Vai al selettore di progetti

  2. Abilita l'API Data Lineage.

    Abilita le API

    Modifiche imminenti alla derivazione dei dati di Spark Consulta le note di rilascio di Managed Service for Apache Spark per l'annuncio di una modifica che renderà automaticamente disponibile la derivazione dei dati di Spark per i tuoi progetti, workload batch e sessioni interattive quando abiliti l'API Data Lineage (consulta Controlla l'importazione della derivazione per un servizio) senza richiedere impostazioni aggiuntive per progetti, workload batch o sessioni interattive.

Ruoli obbligatori

Se il workload batch utilizza il service account Managed Service for Apache Spark predefinito, ha il ruolo Dataproc Worker. Questo ruolo include l'autorizzazione datalineage.locations.processOpenLineageMessage, che il plug-in OpenLineage Spark utilizza per acquisire e pubblicare eventi di derivazione nell'API Data Lineage. Non sono necessari ulteriori interventi.

Se il workload batch o la sessione interattiva utilizza un account di servizio personalizzato, quest'ultimo deve disporre dellaccount di servizio#39;autorizzazione per pubblicare eventi di derivazione. Per seguire il principio del privilegio minimo, concedi il ruolo minimo richiesto per il tuo caso d'uso:

  • Dataproc Worker (roles/dataproc.worker): il ruolo standard richiesto per l'esecuzione del workload di Managed Service for Apache Spark. Poiché questo ruolo include già l'autorizzazione datalineage.locations.processOpenLineageMessage, se il tuo account di servizio personalizzato ha già questo ruolo, non devi concedere un ruolo Data Lineage aggiuntivo.
  • Producer di eventi di derivazione dei dati (roles/datalineage.producer): il ruolo di derivazione dei dati autonomo minimo richiesto per un account di servizio personalizzato per pubblicare eventi di derivazione. Concedi questo ruolo se il account di servizio non dispone del ruolo Dataproc Worker.
  • Data Lineage Editor (roles/datalineage.editor): concede le autorizzazioni per modificare ed eliminare processi ed eventi di derivazione, oltre a pubblicarli. Questo ruolo fornisce più autorizzazioni di quelle necessarie per l'esecuzione standard della pipeline ed è richiesto solo se i workload devono modificare i metadati di derivazione esistenti.
  • Data Lineage Administrator (roles/datalineage.admin): concede l'accesso amministrativo completo alle risorse di derivazione dei dati. Questo ruolo non è richiesto per i service account dei workload.

Abilita la derivazione dei dati di Spark

Puoi attivare la derivazione dei dati di Spark per il tuo progetto, workload batch o sessione interattiva.

Abilita la derivazione dei dati a livello di progetto

Dopo aver abilitato la derivazione dei dati Spark a livello di progetto, i job Spark successivi eseguiti in un workload batch o in una sessione interattiva avranno la derivazione dei dati Spark abilitata.

Per abilitare la derivazione dei dati di Spark nel tuo progetto, imposta i seguenti metadati di progetto personalizzati:

Chiave Valore
DATAPROC_LINEAGE_ENABLED true

Puoi disattivare la derivazione dei dati di Spark per un progetto impostando i metadati DATAPROC_LINEAGE_ENABLED su false.

Abilita la derivazione dei dati su un workload batch Spark

Per abilitare la derivazione dei dati su un workload batch, imposta la proprietà spark.dataproc.lineage.enabled su true quando invii il workload. Questa impostazione esegue l'override di qualsiasi impostazione della derivazione dei dati di Spark a livello di progetto: se la derivazione dei dati di Spark è disabilitata a livello di progetto, ma abilitata per il workload batch, l'impostazione del workload batch ha la precedenza.

Puoi disattivare la derivazione dei dati di Spark su un workload batch di Spark impostando la proprietà spark.dataproc.lineage.enabled su false quando invii il workload.

Questo esempio utilizza gcloud CLI per inviare un workload batch lineage-example.py con la derivazione Spark abilitata.

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

Il seguente codice lineage-example.py legge i dati da una tabella BigQuery pubblica, quindi scrive l'output in una nuova tabella in un set di dati BigQuery esistente. Utilizza un bucket Cloud Storage per l'archiviazione temporanea.

#!/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()

Sostituisci quanto segue:

  • REGION: la regione in cui eseguire il workload
  • BUCKET: il nome di un bucket Cloud Storage esistente in cui archiviare le dipendenze
  • PROJECT_ID, DATASET e TABLE: l'ID progetto, il nome di un set di dati BigQuery esistente e il nome di una nuova tabella da creare nel set di dati (la tabella non deve esistere)

Puoi visualizzare il grafico della derivazione nell'interfaccia utente di Knowledge Catalog.

Grafico di derivazione Spark

Abilita la derivazione dei dati in una sessione interattiva o in un profilo di runtime di Spark (modello di sessione)

Per abilitare la derivazione dei dati in una sessione interattiva o un profilo di runtime Spark (template di sessione), imposta la proprietà spark.dataproc.lineage.enabled su true quando crei la sessione o il profilo di runtime (template di sessione). Ad esempio, nella pagina Crea un profilo di runtime della console Google Cloud , in Configurazione aggiuntiva, modifica Altre personalizzazioni e aggiungi la proprietà spark.dataproc.lineage.enabled con il valore true in Proprietà Spark. Questa impostazione esegue l'override di qualsiasi impostazione della derivazione dei dati di Spark a livello di progetto: se la derivazione dei dati di Spark è disabilitata a livello di progetto, ma abilitata per la sessione interattiva, l'impostazione della sessione interattiva ha la precedenza.

Puoi disattivare la derivazione dei dati di Spark in una sessione interattiva o un profilo di runtime (modello di sessione) di Spark impostando la proprietà spark.dataproc.lineage.enabled su false quando crei la sessione interattiva o il profilo di runtime (modello di sessione).

Il seguente codice del notebook PySpark configura una sessione interattiva di Managed Service for Apache Spark con la derivazione dei dati di Spark abilitata. Quindi, crea una sessione Spark Connect che esegue una query di conteggio parole su un set di dati Shakespeare pubblico di BigQuery, quindi scrive l'output in una nuova tabella in un set di dati BigQuery esistente (consulta Utilizza il client Spark Connect).

# 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.managed_spark_connect import ManagedSparkSession

spark = (
    ManagedSparkSession.builder
    .sessionId("lineage-example")
    .sessionConfig(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()

Sostituisci quanto segue:

  • PROJECT_ID, DATASET e TABLE: l'ID progetto, il nome di un set di dati BigQuery esistente e il nome di una nuova tabella da creare nel set di dati (la tabella non deve esistere)

Puoi visualizzare il grafico della derivazione dei dati facendo clic sul nome della tabella di destinazione elencato nel riquadro di navigazione nella pagina Explorer di BigQuery, quindi selezionando la scheda della derivazione nel riquadro dei dettagli della tabella.

Grafico di derivazione Spark

Visualizza la derivazione in Knowledge Catalog

Un grafico di derivazione mostra le relazioni tra le risorse del progetto e i processi che le hanno create. Puoi visualizzare le informazioni sulla derivazione dei dati nella console Google Cloud o recuperarle dall'API Data Lineage come dati JSON.

Passaggi successivi