Usar o conector do BigQuery com o Dataproc sem servidor para Spark

Use o spark-bigquery-connector com o Apache Spark para ler e gravar dados no e a partir do BigQuery. Este tutorial demonstra um aplicativo PySpark que usa a spark-bigquery-connector:

Usar o conector do BigQuery com sua carga de trabalho

Consulte Dataproc Serverless for Spark releases para determinar a versão do conector do BigQuery instalada na versão do ambiente de execução da carga de trabalho em lote. Se o conector não estiver listado, consulte a próxima seção para instruções sobre como disponibilizar o conector para aplicativos conteinerizados.

Como usar o conector com o ambiente de execução do Spark versão 2.0

O conector do BigQuery não está instalado na versão 2.0 do ambiente de execução do Spark. Ao usar A versão 2.0 do ambiente de execução do Spark, é possível disponibilizar o conector para o aplicativo de uma das seguintes maneiras:

  • Use o parâmetro jars para apontar para um arquivo jar de conector ao enviar sua carga de trabalho em lote do Dataproc Serverless para Spark. O exemplo a seguir especifica um arquivo jar de conector. Consulte o repositório GoogleCloudDataproc/spark-bigquery-connector no GitHub para conferir uma lista de arquivos jar de conector disponíveis.
    • Exemplo da Google Cloud CLI: 
      gcloud dataproc batches submit pyspark \
    --region=region \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.13-version.jar \
    ... other args
  • Inclua o arquivo jar do conector no aplicativo Spark como uma dependência. Consulte Como compilar com o conector.

Calcular custos

Neste tutorial, há componentes faturáveis do Google Cloud, entre eles:

  • Dataproc sem servidor
  • BigQuery
  • Cloud Storage

Use a Calculadora de preços para gerar uma estimativa de custo com base no uso previsto. É possível que novos usuários do Cloud Platform tenham direito a uma avaliação gratuita.

E/S do BigQuery

Este exemplo lê dados do BigQuery em um DataFrame do Spark para executar uma contagem de palavras usando a API de origem de dados padrão.

O conector grava a saída de contagem de palavras no BigQuery da seguinte maneira:

  1. armazenar os dados em buffer em arquivos temporários no bucket do Cloud Storage;

  2. Como copiar os dados em uma operação do bucket do Cloud Storage para o BigQuery

  3. Excluir os arquivos temporários no Cloud Storage após o BigQuery a operação de carregamento do arquivo for concluída (os arquivos temporários também serão excluídos o aplicativo Spark é encerrado). Se a exclusão falhar, será necessário excluir todos os arquivos temporários indesejados do Cloud Storage, que geralmente são colocados em gs://your-bucket/.spark-bigquery-jobid-UUID.

Configurar faturamento

Por padrão. o projeto associado às credenciais ou à conta de serviço é cobrado pelo uso da API. Para faturar um projeto diferente, defina a seguinte configuração: spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

Ele também pode ser adicionado a uma operação de leitura/gravação, da seguinte maneira: .option("parentProject", "<BILLED-GCP-PROJECT>").

Enviar uma carga de trabalho de contagem de palavras do PySpark

  1. Crie o wordcount_dataset com a ferramenta de linha de comando bq em um terminal local ou no Cloud Shell. 
    bq mk wordcount_dataset
  2. Crie um bucket do Cloud Storage com a CLI do Google Cloud em um terminal local ou no Cloud Shell. 
    gcloud storage buckets create gs://your-bucket
  3. Examine o código. 
    #!/usr/bin/python
"""BigQuery I/O PySpark example."""
from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the connector.
bucket = "[your-bucket-name]"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .option('table', 'bigquery-public-data:samples.shakespeare') \
  .load()
words.createOrReplaceTempView('words')

# Perform word count.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')
word_count.show()
word_count.printSchema()

# Saving the data to BigQuery
word_count.write.format('bigquery') \
  .option('table', 'wordcount_dataset.wordcount_output') \
  .save()
  4. Crie wordcount.py localmente em um editor de texto copiando o código PySpark da lista de códigos do PySpark. Substitua o marcador de posição [seu-bucket] pelo nome do bucket do Cloud Storage criado.
  5. Envie a carga de trabalho em lote do PySpark: 
    gcloud dataproc batches submit pyspark wordcount.py \
    --region=region \
    --deps-bucket=your-bucket
    Exemplo de saída do terminal: 
    ...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

    Para visualizar a tabela de saída no console do Google Cloud, abra o BigQuery selecione a tabela wordcount_output e clique em Visualização.

Para mais informações