Utilizzare il connettore BigQuery con Spark

Il connettore spark-bigquery-connector viene utilizzato con Apache Spark per leggere e scrivere dati da e in BigQuery. Questo tutorial fornisce un codice di esempio che utilizza il connettore spark-bigquery-connector in un'applicazione Spark. Per istruzioni sulla creazione di un cluster, consulta Guide rapide di Dataproc.

Il connettore spark-bigquery-sfrutta i BigQuery API Storage durante la lettura dei dati da BigQuery.

Rendi disponibile il connettore per la tua applicazione

Puoi rendere disponibile il connettore Spark-bigquery per la tua applicazione in uno dei seguenti modi:

1. Installa spark-bigquery-connector nella directory Spark jars di ogni utilizzando il nodo Azione di inizializzazione dei connettori Dataproc quando crei il cluster.

2. Fornisci l'URI del connettore quando invii il job:

   1. Console Google Cloud: utilizza l'elemento del job Spark Jars files su nella pagina Invia un job di Dataproc.
   2. gcloud CLI: utilizza il flag gcloud dataproc jobs submit spark --jars.
   3. API Dataproc: utilizza il metodo Campo SparkJob.jarFileUris.

3. Includi il jar nell'applicazione Scala o Java Spark come dipendenza (vedi Compilazione rispetto al connettore).

Nota: se il connettore non è disponibile in fase di runtime, viene visualizzato un ClassNotFoundException.

Come specificare l'URI del jar del connettore

Le versioni del connettore Spark-BigQuery sono elencate nel file GitHub Repository GoogleCloudDataproc/spark-bigquery-connector.

Specifica il jar del connettore sostituendo la scala e la versione del connettore nella seguente stringa URI:

gs://spark-lib/bigquery/spark-bigquery-with-dependencies_SCALA_VERSION-CONNECTOR_VERSION.jar

• Utilizza Scala 2.12 con le versioni immagine di Dataproc 1.5+

  gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-CONNECTOR_VERSION.jar
  

  Esempio della gcloud CLI:

  gcloud dataproc jobs submit spark \
      --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.23.2.jar \
      -- job-args
  

• Utilizza Scala 2.11 con le versioni immagine di Dataproc 1.4 e precedenti:

  gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.11-CONNECTOR_VERSION.jar
  

  Esempio della gcloud CLI:

  gcloud dataproc jobs submit spark \
      --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.11-0.23.2.jar \
      -- job-args
  

Per l'uso non di produzione, puoi anche puntare ai jar latest in questo modo:

1. Versione immagine Dataproc 1.5+: --jars=gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar
2. Immagine Dataproc versione 1.4 e precedenti: --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar

Calcolo dei costi

In questo documento vengono utilizzati i seguenti componenti fatturabili di Google Cloud:

• Dataproc
• BigQuery
• Cloud Storage

Per generare una stima dei costi basata sull'utilizzo previsto, utilizza il Calcolatore prezzi. I nuovi utenti di Google Cloud potrebbero essere idonei per una prova gratuita.

Lettura e scrittura di dati da BigQuery

Questo esempio legge i dati da BigQuery in un DataFrame Spark per eseguire un conteggio parole utilizzando l'origine dati standard all'API.

Il connettore scrive i dati in BigQuery eseguendo prima il buffering di tutti i dati in una tabella temporanea di Cloud Storage. Poi consente di copiare tutti i dati da BigQuery in un'unica operazione. La tenta di eliminare i file temporanei una volta che dell'operazione di caricamento è riuscita e di nuovo al termine dell'applicazione Spark. Se il job non riesce, rimuovi eventuali istanze temporanee rimanenti di archiviazione dei file di Cloud Storage. In genere, BigQuery temporaneo file si trovano in gs://[bucket]/.spark-bigquery-[jobid]-[UUID].

Configurazione della fatturazione

Per impostazione predefinita, il progetto associato alle credenziali o all'account di servizio addebitati in base all'utilizzo dell'API. Per fatturare un progetto diverso, imposta quanto segue configurazione: spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

Può anche essere aggiunto a un'operazione di lettura/scrittura, come segue: .option("parentProject", "<BILLED-GCP-PROJECT>").

Esecuzione del codice

Prima di eseguire questo esempio, crea un set di dati denominato "wordcount_dataset" o cambiare il set di dati di output nel codice con un set di dati BigQuery esistente nel progetto Google Cloud.

Utilizza la il comando bq per creare wordcount_dataset:

bq mk wordcount_dataset

Utilizza il comando Google Cloud CLI per creare un bucket Cloud Storage, che verrà utilizzato per l'esportazione BigQuery:

gcloud storage buckets create gs://[bucket]

Scala

1. Esamina il codice e sostituisci il segnaposto [bucket] con nel bucket Cloud Storage che hai creato in precedenza.

   /*
    * Remove comment if you are not running in spark-shell.
    *
   import org.apache.spark.sql.SparkSession
   val spark = SparkSession.builder()
     .appName("spark-bigquery-demo")
     .getOrCreate()
   */
   
   // Use the Cloud Storage bucket for temporary BigQuery export data used
   // by the connector.
   val bucket = "[bucket]"
   spark.conf.set("temporaryGcsBucket", bucket)
   
   // Load data in from BigQuery. See
   // https://github.com/GoogleCloudDataproc/spark-bigquery-connector/tree/0.17.3#properties
   // for option information.
   val wordsDF =
     (spark.read.format("bigquery")
     .option("table","bigquery-public-data:samples.shakespeare")
     .load()
     .cache())
   
   wordsDF.createOrReplaceTempView("words")
   
   // Perform word count.
   val wordCountDF = spark.sql(
     "SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word")
   wordCountDF.show()
   wordCountDF.printSchema()
   
   // Saving the data to BigQuery.
   (wordCountDF.write.format("bigquery")
     .option("table","wordcount_dataset.wordcount_output")
     .save())
   
   

2. Esegui il codice sul tuo cluster
   1. Utilizza SSH per connetterti al nodo master del cluster Dataproc
      1. Vai alla sezione Cluster di Dataproc nella console Google Cloud, quindi fai clic sul nome del cluster

         

      2. Nella pagina >Dettagli cluster, seleziona la scheda Istanze VM. Quindi, fai clic su SSH a destra del nome del nodo master del cluster

         

         
         Si apre una finestra del browser nella home directory sul nodo master

             Connected, host fingerprint: ssh-rsa 2048 ...
             ...
             user@clusterName-m:~$
             

   2. Crea wordcount.scala con l'elemento vi preinstallato, vim o nano, quindi incolla il testo in Scala il codice Elenco di codice Scala

      nano wordcount.scala
        

   3. Lancia il REPL spark-shell.

      $ spark-shell --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar
      ...
      Using Scala version ...
      Type in expressions to have them evaluated.
      Type :help for more information.
      ...
      Spark context available as sc.
      ...
      SQL context available as sqlContext.
      scala>
      

   4. Esegui wordcount.scala con il comando :load wordcount.scala per creare la tabella wordcount_output di BigQuery. L'output l'elenco mostra 20 righe dall'output del conteggio parole.

      :load wordcount.scala
      ...
      +---------+----------+
      |     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)
      

      
      Per visualizzare l'anteprima della tabella di output, apri la BigQuery seleziona la tabella wordcount_output e poi fai clic su Anteprima.

      

PySpark

1. Esamina il codice e sostituisci il segnaposto [bucket] con nel bucket Cloud Storage che hai creato in precedenza.

   #!/usr/bin/env python
   
   """BigQuery I/O PySpark example."""
   
   from pyspark.sql import SparkSession
   
   spark = SparkSession \
     .builder \
     .master('yarn') \
     .appName('spark-bigquery-demo') \
     .getOrCreate()
   
   # Use the Cloud Storage bucket for temporary BigQuery export data used
   # by the connector.
   bucket = "[bucket]"
   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()
   
   # Save the data to BigQuery
   word_count.write.format('bigquery') \
     .option('table', 'wordcount_dataset.wordcount_output') \
     .save()
   

2. Esegui il codice sul tuo cluster
   Utilizza Dataproc per inviare il codice PySpark: Anziché eseguire manualmente il codice PySpark dall'istanza master del cluster, puoi inviare il file PySpark direttamente al tuo cluster (vedi il Guide rapide di Dataproc). Ecco i passaggi da seguire utilizzando Google Cloud CLI:
   

   1. Crea wordcount.py localmente in un editor di testo copiando il codice PySpark Scheda di codice PySpark
   2. Esegui il codice PySpark inviando il job al cluster con Comando gcloud dataproc jobs submit:

      gcloud dataproc jobs submit pyspark wordcount.py \
          --cluster=cluster-name \
          --region=region \
          --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar
      

   1. Utilizza SSH per connetterti al nodo master del cluster Dataproc
      1. Vai alla sezione Cluster di Dataproc nella console Google Cloud, quindi fai clic sul nome del cluster

         

      2. Nella pagina Dettagli cluster, seleziona la scheda Istanze VM. Quindi, fai clic su SSH a destra del nome del nodo master del cluster

         

         
         Si apre una finestra del browser nella home directory sul nodo master

             Connected, host fingerprint: ssh-rsa 2048 ...
             ...
             user@clusterName-m:~$
             

   2. Crea wordcount.py con l'elemento vi preinstallato, Editor di testo vim o nano, poi incollalo in PySpark il codice Scheda di codice PySpark

      nano wordcount.py
      

   3. Esegui conteggio parole con spark-submit per creare BigQuery Tabella wordcount_output. L'elenco di output mostra 20 righe dall'output del conteggio parole.

      spark-submit --jars gs://spark-lib/bigquery/spark-bigquery-latest.jar wordcount.py
      ...
      +---------+----------+
      |     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)
      

      
      Per visualizzare l'anteprima della tabella di output, apri la BigQuery seleziona la tabella wordcount_output e poi fai clic su Anteprima.

      

Per ulteriori informazioni

• Archiviazione e archiviazione BigQuery Spark SQL - Python
• Creazione di un file di definizione della tabella per un'origine dati esterna
• Esecuzione di query su dati partizionati esternamente
• Suggerimenti per l'ottimizzazione dei job Spark