Carregar dados no BigQuery

Este documento mostra como usar o Serviço Gerenciado para Apache Spark para executar um job do Spark que carrega dados processados do Cloud Storage em uma tabela do BigQuery. O Serviço Gerenciado para Apache Spark simplifica esse processo gerenciando o ambiente do Spark e os conectores necessários.

Antes de começar

  1. Faça login na sua Google Cloud conta do. Se você não conhece o Google Cloud, crie uma conta para avaliar o desempenho dos nossos produtos em cenários reais. Clientes novos também recebem US $300 em créditos para executar, testar e implantar cargas de trabalho.

  2. 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.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. 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

  5. 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.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. 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

    8.
  8. Crie um bucket do Cloud Storage.
  9. Crie um cluster do Serviço Gerenciado para Apache Spark que use a versão de imagem 2.1 ou mais recente.
  10. 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 Google Cloud ID do projeto.

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

Enviar o job do Serviço Gerenciado para Apache Spark

Envie o script do PySpark como um job para o cluster do Serviço Gerenciado para 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 Serviço Gerenciado para Apache Spark.
    • REGION: a região em que o cluster está localizado.

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

Verificar o carregamento de dados

  1. No Google Cloud console do, acesse a página Jobs do Serviço Gerenciado para 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 o projeto e o conjunto de dados e selecione a tabela corpus_word_counts.

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

Como o conector do Spark-BigQuery funciona

O conector do 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 Serviço Gerenciado 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 desvincula a computação do Spark da ingestão do BigQuery. Essa desvinculação permite que cada serviço opere com eficiência e garante alta capacidade de processamento para grandes cargas de dados.

A seguir