Habilita el linaje de datos de Spark en Dataproc

En este documento, se describe cómo habilitar el linaje de datos para tus trabajos de Spark de Dataproc a nivel del proyecto o del clúster.

El linaje de datos es una función de Dataplex que te permite hacer un seguimiento de cómo los datos se mueven a través de tus sistemas: de dónde provienen, a dónde se pasan y qué transformaciones se aplican a ellos.

El linaje de datos está disponible para todos los trabajos de Spark de Dataproc, excepto para los trabajos de transmisión de SparkR y Spark, y admite fuentes de datos de BigQuery y Cloud Storage. Se incluye con Dataproc en las versiones de imágenes 2.0.74 y posteriores, y 2.1.22 y posteriores de Compute Engine.

Una vez que habilites la función en tu clúster de Dataproc, los trabajos de Spark de Dataproc capturarán los eventos del linaje de datos y los publicarán en la API de Data Lineage de Dataplex. Dataproc se integra con la API de Data Lineage a través de OpenLineage con el complemento OpenLineage Spark.

Puedes acceder a la información del linaje de datos a través de Dataplex con lo siguiente:

Antes de comenzar

  1. En la página del selector de proyectos de la consola de Google Cloud, selecciona el proyecto que contiene el clúster de Dataproc para el que deseas realizar un seguimiento del linaje.

    Ir al selector de proyectos

  2. Habilita las APIs de Data Lineage y Dataplex.

    Habilitar las API

Roles obligatorios

Para obtener los permisos que necesitas para usar el linaje de datos en Dataproc, pídele a tu administrador que te otorgue los siguientes roles de IAM en la cuenta de servicio de VM del clúster de Dataproc:

Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

También puedes obtener los permisos necesarios mediante roles personalizados o cualquier otro rol predefinido.

Habilita el linaje de datos de Spark a nivel del proyecto

Puedes habilitar el linaje de datos de Spark a nivel del proyecto. Los trabajos de Spark compatibles que se ejecutan en clústeres creados después de que se habilita el linaje de datos en un proyecto tendrán habilitado el linaje de datos. Ten en cuenta que los trabajos que se ejecutan en clústeres existentes (clústeres que se crearon antes de habilitar el linaje de datos a nivel del proyecto) no tendrán habilitado el linaje de datos.

Cómo habilitar el linaje de datos de Spark a nivel del proyecto

Para habilitar el linaje de datos de Spark a nivel del proyecto, configura los siguientes metadatos personalizados del proyecto:

Clave Valor
DATAPROC_LINEAGE_ENABLED true
DATAPROC_CLUSTER_SCOPES https://www.googleapis.com/auth/cloud-platform

Puedes inhabilitar el linaje de datos de Spark a nivel del proyecto si estableces los metadatos DATAPROC_LINEAGE_ENABLED en false.

Habilita el linaje de datos de Spark a nivel del clúster

Puedes habilitar el linaje de datos de Spark cuando creas un clúster para que todos los trabajos de Spark compatibles que se envíen al clúster tengan habilitado el linaje de datos.

Cómo habilitar el linaje de datos de Spark a nivel del clúster

Para habilitar el linaje de datos de Spark en un clúster, crea un clúster de Dataproc con la propiedad del clúster dataproc:dataproc.lineage.enabled establecida en true.

Clústeres de la versión 2.0 de la imagen: Se requiere el alcance cloud-platform de acceso a la VM del clúster de Dataproc para el linaje de datos de Spark. Los clústeres de la versión de imagen de Dataproc creados con la versión 2.1 de la imagen y versiones posteriores tienen habilitada la cloud-platform. Si especificas la versión de imagen de Dataproc 2.0 cuando creas un clúster, establece el alcance en cloud-platform.

Ejemplo de la CLI de gcloud:

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

Inhabilita el linaje de datos de Spark en un trabajo

Si habilitas el linaje de datos de Spark a nivel del clúster, puedes inhabilitarlo en un trabajo específico pasando la propiedad spark.extraListeners con un valor vacío ("") cuando envíes el trabajo.

Una vez habilitado, no puedes inhabilitar el linaje de datos de Spark en el clúster. Para eliminar el linaje de datos de Spark en todos los trabajos del clúster, puedes volver a crear el clúster sin la propiedad dataproc:dataproc.lineage.enabled.

Enviar un trabajo de Spark

Cuando envías un trabajo de Spark en un clúster de Dataproc que se creó con el linaje de datos de Spark habilitado, Dataproc captura y informa la información del linaje de datos a la API de 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

Notas:

  • Es opcional agregar las propiedades spark.openlineage.namespace y spark.openlineage.appName, que se usan para identificar de forma única el trabajo. Si no agregas estas propiedades, Dataproc usará los siguientes valores predeterminados:
    • Valor predeterminado para spark.openlineage.namespace: PROJECT_ID
    • Valor predeterminado para spark.openlineage.appName: spark.app.name

Consultar el linaje en Dataplex

Un gráfico de linaje muestra las relaciones entre los recursos de tu proyecto y los procesos que los crearon. Puedes ver la información del linaje de datos en la consola de Google Cloud o recuperarla de la API de Data Lineage en forma de datos JSON.

Ejemplo:

El siguiente trabajo de Spark lee datos de una tabla de BigQuery y los escribe en otra.

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

El trabajo de Spark crea el siguiente gráfico de linaje en la IU de Dataplex:

Gráfico de linaje de muestra

¿Qué sigue?