Carica i dati in BigQuery

Questo documento mostra come utilizzare il servizio gestito per Apache Spark per eseguire un job Spark che carica i dati elaborati da Cloud Storage in una tabella BigQuery. Managed Service per Apache Spark semplifica questo processo gestendo l'ambiente Spark e i connettori necessari.

Prima di iniziare

  1. Accedi al tuo account Google Cloud . Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti senza costi per l'esecuzione, il test e il deployment dei workload.

  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. Crea un bucket Cloud Storage.
  9. Crea un cluster Managed Service per Apache Spark che utilizzi la versione dell'immagine 2.1 o successive.
  10. Crea un set di dati BigQuery.

Prepara lo script PySpark

  1. Crea un file Python locale denominato load_analytics_data.py.

  2. Aggiungi il codice seguente al file. Questo script legge i dati da un percorso Cloud Storage, esegue un'aggregazione e scrive il risultato in 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. Sostituisci i seguenti segnaposto:

    • BUCKET_NAME: il nome del bucket Cloud Storage.
    • PROJECT_ID: il tuo Google Cloud ID progetto.

  4. Carica lo script load_analytics_data.py nel bucket Cloud Storage.

Inviare il job Managed Service per Apache Spark

Invia lo script PySpark come job al tuo cluster Managed Service per Apache Spark.

  1. In un terminale, esegui il 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. Sostituisci i seguenti segnaposto:

    • BUCKET_NAME: il nome del bucket Cloud Storage.
    • CLUSTER_NAME: il nome del tuo cluster Managed Service per Apache Spark.
    • REGION: la regione in cui si trova il cluster.

    Il comando invia un job PySpark al servizio Managed Service per Apache Spark. I worker di Managed Service per Apache Spark recuperano lo script dal percorso Cloud Storage specificato e lo eseguono sul cluster.

Verificare il caricamento dei dati

  1. Nella console Google Cloud , vai alla pagina Job di Managed Service for Apache Spark per monitorare l'esecuzione del job e visualizzare i log di output del driver.

  2. Al termine del job, vai alla pagina BigQuery.

  3. Nel riquadro Explorer, individua il progetto e il set di dati, quindi seleziona la tabella corpus_word_counts.

  4. Fai clic sulla scheda Anteprima per esaminare i dati caricati.

Come funziona il connettore Spark-BigQuery

Il connettore Spark-BigQuery consente alle applicazioni Spark di leggere e scrivere in BigQuery. Nei cluster Managed Service per Apache Spark con versioni immagine 2.1 o successive, il connettore è preinstallato.

Il connettore utilizza un metodo di scrittura indiretta per caricare i dati. Questo metodo sfrutta sia Managed Service per Apache Spark che BigQuery per prestazioni elevate.

  1. Il job Spark sul cluster Managed Service per Apache Spark scrive il DataFrame finale in una posizione temporanea in un bucket Cloud Storage.

  2. Al termine della scrittura in Cloud Storage, il connettore attiva un job di caricamento BigQuery.

  3. BigQuery inserisce i dati dalla posizione temporanea di Cloud Storage nella tabella di destinazione.

Questo approccio indiretto disaccoppia il calcolo Spark dall'importazione BigQuery. Questo disaccoppiamento consente a ogni servizio di operare in modo efficiente e garantisce un throughput elevato per i carichi di dati di grandi dimensioni.

Passaggi successivi