Utilizzare Dataproc Metastore con Dataproc

Questo documento spiega come utilizzare BigQuery Metastore con Dataproc su Compute Engine. Questa connessione ti fornisce un singolo metastore condiviso che funziona su motori software open source, come Apache Spark.

Prima di iniziare

  1. Attiva la fatturazione per il tuo progetto Google Cloud . Scopri come controllare se la fatturazione è attivata in un progetto.

  2. Abilita le API BigQuery e Dataproc.

    Abilita le API

  3. (Facoltativo) Scopri come funziona il metastore BigQuery e perché dovresti utilizzarlo.

Ruoli obbligatori

Per ottenere le autorizzazioni necessarie per utilizzare Spark e Dataproc con BigQuery Metastore come metastore, chiedi all'amministratore di concederti i seguenti ruoli IAM:

Per saperne di più sulla concessione dei ruoli, consulta Gestire l'accesso a progetti, cartelle e organizzazioni.

Potresti anche riuscire a ottenere le autorizzazioni richieste tramite i ruoli personalizzati o altri ruoli predefiniti.

Flusso di lavoro generale

Per utilizzare Dataproc su Compute Engine con il metastore BigQuery, segui questi passaggi generali:

  1. Crea un cluster Dataproc o configurane uno esistente.
  2. Connettiti al motore software open source che preferisci, ad esempio Spark.
  3. Utilizza un file JAR per installare il plug-in del catalogo Apache Iceberg sul cluster.
  4. Crea e gestisci le risorse del metastore BigQuery in base alle tue esigenze, a seconda del motore software open source in uso.
  5. In BigQuery, accedi e utilizza le risorse del metastore BigQuery.

Collega il metastore BigQuery a Spark

Le istruzioni riportate di seguito mostrano come collegare Dataproc al metastore BigQuery utilizzando Spark SQL interattivo.

Scarica il plug-in del catalogo Iceberg

Per collegare il metastore BigQuery a Dataproc e Spark, devi utilizzare il file jar del plug-in del catalogo Iceberg del metastore BigQuery.

Questo file è incluso per impostazione predefinita nella versione 2.2 dell'immagine Dataproc. Se i tuoi cluster Dataproc non hanno accesso diretto a internet, devi scaricare il plug-in e caricarlo in un bucket Cloud Storage a cui il tuo cluster Dataproc può accedere.

Scarica il plug-in del catalogo Apache Iceberg per il metastore BigQuery.

Configura un cluster Dataproc

Prima di connetterti al metastore BigQuery, devi configurare un cluster Dataproc.

A tale scopo, puoi creare un nuovo cluster o utilizzarne uno esistente. Successivamente, puoi utilizzare questo cluster per eseguire Spark SQL interattivo e gestire le risorse del metastore BigQuery.

  • Nella subnet della regione in cui viene creato il cluster deve essere attivo l'accesso privato Google (PGA). Per impostazione predefinita, le VM del cluster Dataproc create con una versione dell'immagine 2.2 (predefinita) o successiva hanno solo indirizzi IP interni. Per consentire alle VM del cluster di comunicare con le API di Google, abilita l'accesso privato Google sulla subnet della rete default (o sul nome della rete specificato dall'utente, se applicabile) nella regione in cui viene creato il cluster.

  • Se vuoi eseguire l'esempio dell'interfaccia web di Zeppelin in questa guida, devi utilizzare o creare un cluster Dataproc con il componente facoltativo di Zeppelin abilitato.

Nuovo cluster

Per creare un nuovo cluster Dataproc, esegui il seguente comando gcloud dataproc clusters create. Questa configurazione contiene le impostazioni necessarie per utilizzare il metastore BigQuery.

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

Sostituisci quanto segue:

  • CLUSTER_NAME: un nome per il tuo cluster Dataproc.
  • PROJECT_ID: l'ID del progetto Google Cloud in cui stai creando il cluster.
  • LOCATION: la regione Google Cloud in cui crei il cluster.

Cluster esistente

Per configurare un cluster esistente, aggiungi al cluster il seguente runtime Spark Iceberg.

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2

Puoi aggiungere il runtime utilizzando una delle seguenti opzioni:

Invia un job Spark

Per inviare un job Spark, utilizza uno dei seguenti metodi:

Interfaccia a riga di comando gcloud

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto Google Cloud che contiene il cluster Dataproc.
  • CLUSTER_NAME: il nome del cluster Dataproc che utilizzi per eseguire il job Spark SQL.
  • REGION: la regione Compute Engine in cui si trova il cluster.
  • LOCATION: la posizione delle risorse BigQuery.
  • CATALOG_NAME: il nome del catalogo Spark che utilizzi con il tuo job SQL.
  • WAREHOUSE_DIRECTORY: la cartella Cloud Storage contenente il data warehouse. Questo valore inizia con gs://.
  • SPARK_SQL_COMMAND: la query Spark SQL che vuoi eseguire. Questa query include i comandi per creare le risorse. Ad esempio, per creare un ambito e una tabella.

Spark interattivo

Connettiti a Spark e installa il plug-in del catalogo

Per installare il plug-in del catalogo per il metastore BigQuery, collega il tuo cluster Dataproc utilizzando SSH.

  1. Nella console Google Cloud , vai alla pagina Istanze VM.

  2. Per connetterti a un'istanza VM Dataproc, fai clic su SSH nell'elenco delle istanze della macchina virtuale. L'output è simile al seguente:

    Connected, host fingerprint: ssh-rsa ...
Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
...
example-cluster@cluster-1-m:~$

  3. Nel terminale, esegui il seguente comando di inizializzazione del metastore BigQuery:

    spark-sql \
--jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar \
--conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
--conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
--conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \
--conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY

    Sostituisci quanto segue:

    • CATALOG_NAME: il nome del catalogo Spark che utilizzi con il tuo job SQL.
    • PROJECT_ID: l'ID progetto Google Cloud del catalogo del metastore BigQuery a cui si collega il catalogo Spark.
    • LOCATION: la posizione Google Cloud del metastore BigQuery.
    • WAREHOUSE_DIRECTORY: la cartella Cloud Storage contenente il data warehouse. Questo valore inizia con gs://.

    Dopo aver eseguito la connessione a un cluster, il terminale Spark visualizza il prompt spark-sql.

    spark-sql (default)>

Gestire le risorse del metastore di BigQuery

Ora sei connesso al metastore BigQuery. Puoi visualizzare le risorse esistenti o crearne di nuove in base ai metadati archiviati nel metastore BigQuery.

Ad esempio, prova a eseguire i seguenti comandi nella sessione interattiva di Spark SQL per creare uno spazio dei nomi e una tabella Iceberg.

  • Utilizza il catalogo personalizzato Iceberg:

    USE `CATALOG_NAME`;

  • Crea uno spazio dei nomi:

    CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

  • Utilizza lo spazio dei nomi creato:

    USE NAMESPACE_NAME;

  • Crea una tabella Iceberg:

    CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

  • Per inserire una riga di tabella:

    INSERT INTO TABLE_NAME VALUES (1, "first row");

  • Aggiungi una colonna della tabella:

    ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

  • Visualizza i metadati della tabella:

    DESCRIBE EXTENDED TABLE_NAME;

  • Elenca le tabelle nello spazio dei nomi:

    SHOW TABLES;

Notebook Zeppelin

  1. Nella console Google Cloud , vai alla pagina Cluster Dataproc.

    Vai ai cluster Dataproc

  2. Fai clic sul nome del cluster che vuoi utilizzare.

    Viene visualizzata la pagina Dettagli cluster.

  3. Nel menu di navigazione, fai clic su Interfacce web.

  4. In Gateway dei componenti, fai clic su Zeppelin. Viene visualizzata la pagina del notebook Zeppelin.

  5. Nel menu di navigazione, fai clic su Notebook e poi su +Crea nuova nota.

  6. Nella finestra di dialogo, inserisci il nome di un notebook. Lascia selezionato Spark come interprete predefinito.

  7. Fai clic su Crea. Viene creato un nuovo notebook.

  8. Nel notebook, fai clic sul menu delle impostazioni e poi su Interprete.

  9. Nel campo Cerca interpreti, cerca Spark.

  10. Fai clic su Modifica.

  11. Nel campo Spark.jars, inserisci l'URI del jar Spark.

    https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar

  12. Fai clic su Salva.

  13. Fai clic su OK.

  14. Copia il seguente codice PySpark nel tuo notebook Zeppelin.

    %pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("BigQuery Metastore Iceberg") \
.config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION") \
.config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \
.getOrCreate()
spark.sql("select version()").show()
spark.sql("USE `CATALOG_NAME`;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE TABLE_NAME;").show()

    Sostituisci quanto segue:

    • CATALOG_NAME: il nome del catalogo Spark da utilizzare per il job SQL.
    • PROJECT_ID: l'ID del progetto Google Cloud che contiene il cluster Dataproc.
    • WAREHOUSE_DIRECTORY: la cartella Cloud Storage contenente il data warehouse. Questo valore inizia con gs://.
    • NAMESPACE_NAME: il nome dello spazio dei nomi che fa riferimento alla tabella Spark.
    • WAREHOUSE_DIRECTORY: l'URI della cartella Cloud Storage in cui è archiviato il data warehouse.
    • TABLE_NAME: un nome per la tabella Spark.

  15. Fai clic sull'icona di esecuzione o premi Shift-Enter per eseguire il codice. Al termine del job, viene visualizzato il messaggio di stato "Job Spark completato" e l'output mostra i contenuti della tabella:

Passaggi successivi