"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Usa el linaje de datos con el Servicio administrado para Apache Spark

En este documento, se describe cómo habilitar el linaje de datos en las cargas de trabajo por lotes y las sesiones interactivas de Managed Service para Apache Spark a nivel de proyecto, carga de trabajo por lotes o sesión interactiva o perfil de entorno de ejecución (plantilla de sesión).

Descripción general

El linaje de datos es una función de Knowledge Catalog que te permite supervisar el movimiento de los datos a través de tus sistemas para saber de dónde provienen, a dónde se pasan y qué transformaciones se les aplican.

Las cargas de trabajo y las sesiones de Managed Service para Apache Spark capturan eventos de linaje y los publican en la API de Data Lineage de Knowledge Catalog. Managed Service para Apache Spark se integra en la API de Data Lineage a través de OpenLineage con el complemento de OpenLineage Spark.

Puedes acceder a la información del linaje a través de Knowledge Catalog con gráficos de linaje y la API de Data Lineage. Para obtener más información, consulta Visualiza gráficos de linaje en Knowledge Catalog.

Disponibilidad

El linaje de datos, que admite fuentes de datos de BigQuery y Cloud Storage, está disponible para las cargas de trabajo y las sesiones que se ejecutan con las versiones de entorno de ejecución de Managed Service para Apache Spark compatibles, con las siguientes excepciones y limitaciones:

El linaje de datos no está disponible para las cargas de trabajo o sesiones de SparkR o Spark Streaming.

Antes de comenzar

En la página del selector de proyectos de la consola de Google Cloud , selecciona el proyecto que usarás para tus cargas de trabajo o sesiones de Managed Service para Apache Spark.

Ir al selector de proyectos
Habilita la API de Data Lineage.

Habilitar las API

Próximos cambios en el linaje de datos de Spark Consulta las notas de la versión de Managed Service para Apache Spark para conocer el anuncio de un cambio que hará que el linaje de datos de Spark esté disponible automáticamente para tus proyectos, cargas de trabajo por lotes y sesiones interactivas cuando habilites la API de Data Lineage (consulta Controla la transferencia de linaje para un servicio) sin necesidad de parámetros de configuración adicionales para el proyecto, la carga de trabajo por lotes o la sesión interactiva.

Roles obligatorios

Si tu carga de trabajo por lotes usa la cuenta de servicio predeterminada de Managed Service para Apache Spark, tiene el rol Dataproc Worker. Este rol incluye el permiso datalineage.locations.processOpenLineageMessage, que el complemento de OpenLineage Spark usa para capturar y publicar eventos de linaje en la API de Data Lineage. No es necesario realizar ninguna otra acción.

Si tu carga de trabajo por lotes o sesión interactiva usa una cuenta de servicio personalizada, esta debe tener permiso para publicar eventos de linaje. Para seguir el principio de privilegio mínimo, otorga el rol mínimo requerido para tu caso de uso:

Trabajador de Dataproc (roles/dataproc.worker): Es el rol estándar necesario para la ejecución de cargas de trabajo de Managed Service para Apache Spark. Debido a que este rol ya incluye el permiso datalineage.locations.processOpenLineageMessage, si tu cuenta de servicio personalizada ya tiene este rol, no necesitas otorgar un rol de linaje de datos adicional.
Productor de eventos de linaje de datos (roles/datalineage.producer): Es el rol mínimo independiente de linaje de datos que se requiere para que una cuenta de servicio personalizada publique eventos de linaje. Otorga este rol si la cuenta de servicio no tiene el rol de trabajador de Dataproc.
Editor de linaje de datos (roles/datalineage.editor): Otorga permisos para editar y borrar procesos y eventos de linaje, además de publicarlos. Este rol proporciona más permisos de los necesarios para la ejecución de la canalización estándar y solo se requiere si las cargas de trabajo deben modificar los metadatos de linaje existentes.
Administrador de linaje de datos (roles/datalineage.admin): otorga acceso de administrador completo a los recursos de linaje de datos. Este rol no es obligatorio para las cuentas de servicio de las cargas de trabajo.

Habilita el linaje de datos de Spark

Puedes habilitar el linaje de datos de Spark para tu proyecto, carga de trabajo por lotes o sesión interactiva.

Habilita el linaje de datos a nivel del proyecto

Después de habilitar el linaje de datos de Spark a nivel del proyecto, los trabajos de Spark posteriores que se ejecuten en una carga de trabajo por lotes o una sesión interactiva tendrán habilitado el linaje de datos de Spark.

Para habilitar el linaje de datos de Spark en tu proyecto, establece los siguientes metadatos personalizados del proyecto:

Clave	Valor
`DATAPROC_LINEAGE_ENABLED`	`true`

Puedes inhabilitar el linaje de datos de Spark para un proyecto si configuras los metadatos DATAPROC_LINEAGE_ENABLED como false.

Habilita el linaje de datos en una carga de trabajo por lotes de Spark

Para habilitar el linaje de datos en una carga de trabajo por lotes, establece la propiedad spark.dataproc.lineage.enabled en true cuando envíes la carga de trabajo. Este parámetro de configuración anula cualquier parámetro de configuración del linaje de datos de Spark a nivel de proyecto: si el linaje de datos de Spark está inhabilitado a nivel de proyecto, pero habilitado para la carga de trabajo por lotes, el parámetro de configuración de la carga de trabajo por lotes tiene prioridad.

Puedes inhabilitar el linaje de datos de Spark en una carga de trabajo por lotes de Spark estableciendo la propiedad spark.dataproc.lineage.enabled en false cuando envías la carga de trabajo.

En este ejemplo, se usa gcloud CLI para enviar una carga de trabajo por lotes de lineage-example.py con el linaje de Spark habilitado.

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

El siguiente código de lineage-example.py lee datos de una tabla pública de BigQuery y, luego, escribe el resultado en una tabla nueva en un conjunto de datos existente de BigQuery. Usa un bucket de Cloud Storage para el almacenamiento temporal.

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

Reemplaza lo siguiente:

REGION: Es la región en la que se ejecutará la carga de trabajo.
BUCKET: Es el nombre de un bucket de Cloud Storage existente para almacenar dependencias.
PROJECT_ID, DATASET y TABLE: El ID del proyecto, el nombre de un conjunto de datos de BigQuery existente y el nombre de una tabla nueva que se creará en el conjunto de datos (la tabla no debe existir)

Puedes ver el gráfico de linaje en la IU de Knowledge Catalog.

Gráfico de linaje de Spark

Habilita el linaje de datos en una sesión interactiva o un perfil de entorno de ejecución de Spark (plantilla de sesión)

Para habilitar el linaje de datos en una sesión interactiva de Spark o un perfil de entorno de ejecución (plantilla de sesión), establece la propiedad spark.dataproc.lineage.enabled en true cuando crees la sesión o el perfil de entorno de ejecución (plantilla de sesión). Por ejemplo, en la página Crea un perfil de entorno de ejecución de la consola de Google Cloud , en Configuración adicional, edita Otras personalizaciones y agrega la propiedad spark.dataproc.lineage.enabled con el valor true en Propiedades de Spark. Este parámetro de configuración anula cualquier parámetro de configuración del linaje de datos de Spark a nivel de proyecto: si el linaje de datos de Spark está inhabilitado a nivel de proyecto, pero habilitado para la sesión interactiva, el parámetro de configuración de la sesión interactiva tiene prioridad.

Puedes inhabilitar el linaje de datos de Spark en una sesión interactiva o un perfil de entorno de ejecución (plantilla de sesión) de Spark configurando la propiedad spark.dataproc.lineage.enabled como false cuando creas la sesión interactiva o el perfil de entorno de ejecución (plantilla de sesión).

El siguiente código de notebook de PySpark configura una sesión interactiva de Managed Service para Apache Spark con el linaje de datos de Spark habilitado. Luego, crea una sesión de Spark Connect que ejecuta una consulta de recuento de palabras en un conjunto de datos público de Shakespeare en BigQuery y, luego, escribe el resultado en una tabla nueva en un conjunto de datos de BigQuery existente (consulta Usa el cliente de 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()