Ative a linhagem de dados do Spark no Dataproc

Este documento descreve como ativar a linhagem de dados para as suas tarefas do Dataproc Spark ao nível do projeto ou do cluster.

A linhagem de dados é uma funcionalidade do catálogo universal do Dataplex que lhe permite acompanhar a forma como os dados se movem nos seus sistemas: de onde vêm, para onde são transmitidos e que transformações lhes são aplicadas.

A linhagem de dados está disponível para todas as tarefas do Dataproc Spark, exceto as tarefas do SparkR e do Spark Streaming, e suporta origens de dados do BigQuery e do Cloud Storage. Está incluído no Dataproc no Compute Engine 2.0.74+, 2.1.22+, 2.2.50 e versões de imagem posteriores.

Depois de ativar a funcionalidade no cluster do Dataproc, as tarefas do Spark do Dataproc capturam eventos de linhagem de dados e publicam-nos na API Data Lineage do catálogo universal do Dataplex. O Dataproc integra-se com a API Data Lineage através do OpenLineage, usando o plug-in OpenLineage Spark.

Pode aceder às informações de linhagem de dados através do catálogo universal do Dataplex, usando o seguinte:

Antes de começar

  1. Na Google Cloud consola, na página do seletor de projetos, selecione o projeto que contém o cluster do Dataproc para o qual quer acompanhar a linhagem.

    Aceder ao seletor de projetos

  2. Ative a API Data Lineage.

    Ative as APIs

Funções necessárias

Se criar um cluster do Dataproc com a conta de serviço da VM predefinida, tem a função Dataproc Worker que ativa a linhagem de dados. Não é necessária nenhuma ação adicional.

No entanto, se criar um cluster do Dataproc que use uma conta de serviço personalizada, para ativar a linhagem de dados no cluster, tem de conceder uma função necessária à conta de serviço personalizada, conforme explicado no parágrafo seguinte.

Para receber as autorizações de que precisa para usar a linhagem de dados com o Dataproc, peça ao seu administrador que lhe conceda as seguintes funções do IAM na conta de serviço personalizada do cluster:

Para mais informações sobre a atribuição de funções, consulte o artigo Faça a gestão do acesso a projetos, pastas e organizações.

Também pode conseguir as autorizações necessárias através de funções personalizadas ou outras funções predefinidas.

Ative a linhagem de dados do Spark ao nível do projeto

Pode ativar a linhagem de dados do Spark ao nível do projeto. As tarefas do Spark suportadas executadas em clusters criados depois de a linhagem de dados ser ativada num projeto têm a linhagem de dados ativada. Tenha em atenção que os trabalhos executados em clusters existentes, ou seja, clusters criados antes de ativar a linhagem de dados ao nível do projeto, não têm a linhagem de dados ativada.

Ative a linhagem de dados do Spark ao nível do projeto

Para ativar a linhagem de dados do Spark ao nível do projeto, defina os seguintes metadados personalizados do projeto:

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

Pode desativar a linhagem de dados do Spark ao nível do projeto definindo os metadados DATAPROC_LINEAGE_ENABLED como false.

Ative a linhagem de dados do Spark ao nível do cluster

Pode ativar a linhagem de dados do Spark quando cria um cluster para que todas as tarefas do Spark suportadas enviadas para o cluster tenham a linhagem de dados ativada.

Ative a linhagem de dados do Spark ao nível do cluster

Para ativar a linhagem de dados do Spark num cluster, crie um cluster do Dataproc com a propriedade do cluster dataproc:dataproc.lineage.enabled definida como true.

Clusters da versão de imagem 2.0: o acesso à VM do cluster do Dataproc cloud-platform é necessário para a linhagem de dados do Spark. Os clusters da versão 2.1 e posteriores da imagem do Dataproc criados com a imagem têm o cloud-platform ativado. Se especificar a versão da imagem do Dataproc 2.0 quando criar um cluster, defina o âmbito como cloud-platform.

Exemplo da CLI gcloud:

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

Desative a linhagem de dados do Spark numa tarefa

Se ativar a linhagem de dados do Spark ao nível do cluster, pode desativar a linhagem de dados do Spark numa tarefa específica transmitindo a propriedade spark.extraListeners com um valor vazio ("") quando envia a tarefa.

Depois de ativada, não pode desativar a linhagem de dados do Spark no cluster. Para eliminar a linhagem de dados do Spark em todas as tarefas de cluster, pode recriar o cluster sem a propriedade dataproc:dataproc.lineage.enabled.

Envie uma tarefa do Spark

Quando envia uma tarefa do Spark num cluster do Dataproc criado com a linhagem de dados do Spark ativada, o Dataproc captura e comunica as informações de linhagem de dados à API 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:

  • A adição das propriedades spark.openlineage.namespace e spark.openlineage.appName, que são usadas para identificar exclusivamente a tarefa, é opcional. Se não adicionar estas propriedades, o Dataproc usa os seguintes valores predefinidos:
    • Valor predefinido para spark.openlineage.namespace: PROJECT_ID
    • Valor predefinido para spark.openlineage.appName: spark.app.name

Veja a linhagem no catálogo universal do Dataplex

Um gráfico de linhagem apresenta as relações entre os recursos do seu projeto e os processos que os criaram. Pode ver informações de linhagem de dados na Google Cloud consola ou obtê-las a partir da API Data Lineage sob a forma de dados JSON.

Exemplo de código do PySpark:

A seguinte tarefa do PySpark lê dados de uma tabela pública do BigQuery e, em seguida, escreve o resultado numa nova tabela num conjunto de dados do BigQuery existente. Usa um contentor 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()

bucket = 'gs://BUCKET`
spark.conf.set('temporaryCloudStorageBucket', bucket)

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) \
  .save()

Faça as seguintes substituições:

  • BUCKET: o nome de um contentor do Cloud Storage existente.

  • PROJECT_ID, DATASET e TABLE: insira o ID do projeto, o nome de um conjunto de dados do BigQuery existente e o nome de uma nova tabela a criar no conjunto de dados (a tabela não pode existir).

Pode ver o gráfico de linhagem na IU do catálogo universal do Dataplex.

Exemplo de gráfico de linhagem

O que se segue?