Carregar dados no BigQuery

Neste documento, mostramos como usar o serviço gerenciado para Apache Spark e executar um job do Spark que carrega dados processados do Cloud Storage em uma tabela do BigQuery. O Managed Service for Apache Spark simplifica esse processo gerenciando o ambiente do Spark e os conectores necessários.

Consulte Transformar dados para um guia sobre o processamento de dados usando PySpark no serviço gerenciado para Apache Spark.

Antes de começar

Faça login na sua conta do Google Cloud . Se você começou a usar o Google Cloud, crie uma conta para avaliar o desempenho de nossos produtos em situações reais. Clientes novos também recebem US$ 300 em créditos para executar, testar e implantar cargas de trabalho.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigQuery, and Cloud Storage APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigQuery, and Cloud Storage APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

1. Crie um bucket do Cloud Storage.
2. Crie um cluster do Managed Service for Apache Spark que use a versão de imagem 2.1 ou mais recente.
3. Crie um conjunto de dados do BigQuery.

Preparar o script do PySpark

1. Crie um arquivo Python local chamado load_analytics_data.py.

2. Adicione o seguinte código ao arquivo. Esse script lê dados de um caminho do Cloud Storage, realiza uma agregação e grava o resultado no BigQuery.

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import sum as _sum
   import sys
   
   # --- Configuration ---
   gcs_bucket = "BUCKET_NAME"
   bq_project = "PROJECT_ID"
   bq_dataset = "DATASET"
   bq_table = "corpus_word_counts"
   
   # --- Paths ---
   processed_path = f"gs://{gcs_bucket}/processed/processed_data"
   temp_gcs_path = f"{gcs_bucket}"
   
   # --- Spark Session Initialization ---
   spark = SparkSession.builder \
      .appName("Dataproc-BigQuery-Load") \
      .config("spark.hadoop.google.cloud.bigdata.connector.temporary.gcs.bucket", temp_gcs_path) \
      .getOrCreate()
   
   # --- Read Processed Data from Cloud Storage ---
   processed_df = spark.read.parquet(processed_path)
   
   # --- Final Aggregation for Analytics-Ready Stage ---
   analytics_df = processed_df.groupBy("corpus") \
      .agg(_sum("word_count_int").alias("total_word_count")) \
      .orderBy("corpus")
   
   print("Aggregated Analytics-Ready data:")
   analytics_df.show()
   
   # --- Write DataFrame to BigQuery ---
   print(f"Writing data to BigQuery table: {bq_dataset}.{bq_table}")
   
   analytics_df.write \
      .format("bigquery") \
      .option("table", f"{bq_project}.{bq_dataset}.{bq_table}") \
      .mode("append") \
      .save()
   
   print("Successfully wrote data to BigQuery.")
   
   # --- Stop Spark Session ---
   spark.stop()
   

3. Substitua os seguintes marcadores de posição:

   • BUCKET_NAME: o nome do bucket do Cloud Storage.
   • PROJECT_ID: o ID do projeto do Google Cloud.

4. Faça upload do script load_analytics_data.py para o bucket do Cloud Storage.

Enviar o job do Managed Service for Apache Spark

Envie o script PySpark como um job para o cluster do Managed Service for Apache Spark.

1. Em um terminal, execute o comando gcloud dataproc jobs submit pyspark:

   gcloud dataproc jobs submit pyspark gs://YOUR_BUCKET_NAME/scripts/load_analytics_data.py \
       --cluster=CLUSTER_NAME \
       --region=REGION
   

2. Substitua os seguintes marcadores de posição:

   • BUCKET_NAME: o nome do bucket do Cloud Storage.
   • CLUSTER_NAME: o nome do cluster do Managed Service for Apache Spark.
   • REGION: a região em que o cluster está localizado.

   O comando envia um job PySpark para o Managed Service for Apache Spark. Os workers do Managed Service for Apache Spark buscam o script no caminho especificado do Cloud Storage e o executam no cluster.

Verificar o carregamento de dados

1. No console do Google Cloud , acesse a página Jobs do Managed Service for Apache Spark para monitorar a execução do job e conferir os registros de saída do driver.

2. Quando o job for concluído, acesse a página do BigQuery.

3. No painel "Explorer", encontre seu projeto e conjunto de dados e selecione a tabela corpus_word_counts.

4. Clique na guia Visualização para inspecionar os dados carregados.

Como o conector Spark-BigQuery funciona

O conector Spark-BigQuery permite que aplicativos Spark leiam e gravem no BigQuery. Em clusters do Serviço gerenciado para Apache Spark com versões de imagem 2.1 ou mais recentes, o conector é pré-instalado.

O conector usa um método de gravação indireta para carregar dados. Esse método aproveita o Serviço gerenciado para Apache Spark e o BigQuery para alto desempenho.

1. O job do Spark no cluster do Managed Service para Apache Spark grava o DataFrame final em um local temporário em um bucket do Cloud Storage.

2. Depois que a gravação no Cloud Storage é concluída, o conector aciona um job de carregamento do BigQuery.

3. O BigQuery ingere os dados do local temporário do Cloud Storage na tabela de destino.

Essa abordagem indireta desacopla a computação do Spark da ingestão do BigQuery. Esse desacoplamento permite que cada serviço opere com eficiência e garante alta capacidade de processamento para grandes cargas de dados.

A seguir

• Saiba mais sobre o conector Spark-BigQuery.
• Saiba como consultar e visualizar dados no BigQuery.
• Saiba como visualizar dados do BigQuery usando o Looker.