"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 uses AI technology to translate content into your preferred language. AI translations can contain errors.

Usar linhagem de dados com o Managed Service for Apache Spark

Este documento descreve como ativar a linhagem de dados em cargas de trabalho em lote e sessões interativas do Serviço Gerenciado para Apache Spark no nível do projeto, carga de trabalho em lote ou sessão interativa.

Visão geral

A linhagem de dados é um recurso do Knowledge Catalog que permite acompanhar como os dados se movimentam nos sistemas: de onde vêm , para onde vão e quais transformações são aplicadas a eles.

As cargas de trabalho e sessões do Serviço Gerenciado para Apache Spark capturam eventos de linhagem e os publicam em o Knowledge Catalog API Data Lineage. O Serviço Gerenciado para Apache Spark se integra à API Data Lineage pelo OpenLineage, usando o plug-in OpenLineage Spark.

É possível acessar informações de linhagem pelo Knowledge Catalog, usando gráficos de linhagem e a API Data Lineage. Para mais informações, consulte Visualizar gráficos de linhagem no Knowledge Catalog.

Disponibilidade

A linhagem de dados, que oferece suporte a origens de dados do BigQuery e do Cloud Storage está disponível para cargas de trabalho e sessões executadas com versões compatíveis do ambiente de execução do Serviço Gerenciado para Apache Spark com as seguintes exceções e limitações:

A linhagem de dados não está disponível para cargas de trabalho ou sessões de streaming do SparkR ou do Spark.

Antes de começar

Na página do seletor de projetos no Google Cloud console, selecione o projeto a ser usado para as cargas de trabalho ou sessões do Serviço Gerenciado para Apache Spark.

Acessar o seletor de projetos
Ative a API Data Lineage.

Ativar as APIs

Próximas mudanças na linhagem de dados do Spark Consulte as notas de lançamento do Serviço Gerenciado para Apache Spark para conferir o anúncio de uma mudança que disponibilizará automaticamente a linhagem de dados do Spark para seus projetos, cargas de trabalho em lote, e sessões interativas quando você ativar a API Data Lineage (consulte Controlar a ingestão de linhagem de um serviço) sem exigir outras configurações de projeto, carga de trabalho em lote, ou sessão interativa.

Funções exigidas

Se a carga de trabalho em lote usar a conta de serviço padrão do Serviço Gerenciado para Apache Spark, ela terá o papel Managed Service for Apache Spark Worker, que contém as permissões exigidas pela linhagem de dados.

No entanto, se a carga de trabalho em lote usar uma conta de serviço personalizada para ativar a linhagem de dados, você precisará conceder um dos papéis listados no parágrafo a seguir, que contêm as permissões exigidas pela linhagem de dados, à conta de serviço personalizada.

Para receber as permissões necessárias para usar a linhagem de dados com o Serviço Gerenciado para Apache Spark, peça ao administrador para conceder a você os seguintes papéis do IAM na conta de serviço personalizada da carga de trabalho em lote:

Conceda um dos seguintes papéis:
- Worker do Serviço Gerenciado para Apache Spark (roles/dataproc.worker)
- Editor de linhagem de dados (roles/datalineage.editor)
- Produtor de linhagem de dados (roles/datalineage.producer)
- Administrador de linhagem de dados (roles/datalineage.admin)

Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Também é possível conseguir as permissões necessárias usando personalizados papéis ou outros predefinidos papéis.

Ativar a linhagem de dados do Spark

É possível ativar a linhagem de dados do Spark para seu projeto, carga de trabalho em lote ou sessão interativa.

Ativar a linhagem de dados no nível do projeto

Depois de ativar a linhagem de dados do Spark no nível do projeto, os jobs do Spark subsequentes executados em uma carga de trabalho em lote ou sessão interativa terão a linhagem de dados do Spark ativada.

Para ativar a linhagem de dados do Spark no seu projeto, defina os seguintes metadados personalizados do projeto:

Chave	Valor
`DATAPROC_LINEAGE_ENABLED`	`true`

É possível desativar a linhagem de dados do Spark para um projeto definindo os metadados DATAPROC_LINEAGE_ENABLED como false.

Ativar a linhagem de dados em uma carga de trabalho em lote do Spark

Para ativar a linhagem de dados em uma carga de trabalho em lote, defina a propriedade spark.dataproc.lineage.enabled como true ao enviar a carga de trabalho. Essa configuração substitui qualquer configuração de linhagem de dados do Spark no nível do projeto: se a linhagem de dados do Spark estiver desativada no nível do projeto, mas ativada para a carga de trabalho em lote, a configuração da carga de trabalho em lote terá precedência.

É possível desativar a linhagem de dados do Spark em uma carga de trabalho em lote do Spark definindo a propriedade spark.dataproc.lineage.enabled como false ao enviar a carga de trabalho.

Este exemplo usa a CLI gcloud para enviar uma carga de trabalho em lote lineage-example.py com a linhagem do Spark ativada.

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

O código lineage-example.py a seguir lê dados de uma tabela pública do BigQuery e grava a saída em uma nova tabela em um conjunto de dados do BigQuery. Ele usa um bucket do Cloud Storage para armazenamento temporário.

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

Substitua:

REGION: a região em que a carga de trabalho será executada
BUCKET: o nome de um bucket do Cloud Storage para armazenar dependências
PROJECT_ID, DATASET, e TABLE: o ID do projeto, o nome de um conjunto de dados do BigQuery e o nome de uma nova tabela a ser criada no conjunto de dados (a tabela não pode existir)

É possível visualizar o gráfico de linhagem na interface do Knowledge Catalog.

Gráfico de linhagem do Spark

Ativar a linhagem de dados em uma sessão interativa ou modelo de sessão do Spark

Para ativar a linhagem de dados em uma sessão interativa ou modelo de sessão do Spark, defina a propriedade spark.dataproc.lineage.enabled como true ao criar a sessão ou o modelo de sessão. Essa configuração substitui qualquer configuração de linhagem de dados do Spark no nível do projeto: se a linhagem de dados do Spark estiver desativada no nível do projeto, mas ativada para a sessão interativa, a configuração da sessão interativa terá precedência.

É possível desativar a linhagem de dados do Spark em uma sessão interativa ou modelo de sessão do Spark definindo a propriedade spark.dataproc.lineage.enabled como false ao criar a sessão interativa ou o modelo de sessão.

O código de notebook do PySpark a seguir configura uma sessão interativa do Serviço Gerenciado para Apache Spark com a linhagem de dados do Spark ativada. Em seguida, ele cria uma sessão do Spark Connect que executa uma consulta de contagem de palavras em um conjunto de dados público do BigQuery Shakespeare e grava a saída em uma nova tabela em um conjunto de dados do BigQuery existente (consulte Criar uma sessão do Spark em um notebook do 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()