Utilizzare il connettore Bigtable Spark

Il connettore Spark di Bigtable consente di leggere e scrivere dati da e verso Bigtable. Puoi leggere i dati dall'applicazione Spark utilizzando Spark SQL e DataFrames. Il connettore Bigtable Spark supporta le seguenti operazioni di Bigtable:

• Scrittura di dati
• Lettura di dati
• crea una nuova tabella

Questo documento mostra come convertire una tabella Spark SQL DataFrames in una tabella Bigtable e quindi compilare e creare un file JAR per inviare un job Spark.

Stato del supporto di Spark e Scala

Il connettore Spark di Bigtable supporta solo la versione Scala 2.12 e le seguenti versioni di Spark:

• 2.4.8
• 3,2.x
• 3.3.x
• 3,4.x
• 3,5.x

Il connettore Spark di Bigtable supporta le seguenti versioni di Dataproc:

• Cluster con versione immagine 1.5
• Cluster con versione immagine 2.0
• Cluster con versione dell'immagine 2.1
• Cluster con versione dell'immagine 2.2
• Runtime Dataproc serverless versione 1.0

Calcolare i costi

Se decidi di utilizzare uno dei seguenti componenti fatturabili di Google Cloud, ti verranno addebitate le risorse che utilizzi:

• Bigtable (non ti viene addebitato alcun costo per l'uso dell'emulatore Bigtable)
• Dataproc
• Cloud Storage

I prezzi di Dataproc si applicano all'utilizzo di Dataproc nei cluster di Compute Engine. I prezzi di Dataproc Serverless si applicano ai carichi di lavoro e alle sessioni eseguiti su Dataproc Serverless per Spark.

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il Calcolatore prezzi.

Prima di iniziare

Completa i seguenti prerequisiti prima di utilizzare il connettore Bigtable Spark.

Ruoli obbligatori

Per ottenere le autorizzazioni necessarie per utilizzare il connettore Bigtable Spark, chiedi all'amministratore di concederti i seguenti ruoli IAM nel progetto:

• (Facoltativo) Amministratore Bigtable (roles/bigtable.admin): consente di leggere o scrivere dati e creare una nuova tabella.
• Utente Bigtable (roles/bigtable.user): consente di leggere o scrivere dati, ma non di creare una nuova tabella.

Per saperne di più sulla concessione dei ruoli, consulta Gestire l'accesso.

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

Nota: le autorizzazioni associate a bigtable.admin rappresentano il livello più alto di autorizzazioni. Utilizza altri ruoli predefiniti di Bigtable in base al livello di accesso che vuoi concedere.

Se utilizzi Dataproc o Cloud Storage, potrebbero essere necessarie autorizzazioni aggiuntive. Per ulteriori informazioni, vedi Autorizzazioni Dataproc e Cloud Storage.

Configura Spark

Oltre a creare un'istanza Bigtable, devi configurare l'istanza Spark. Puoi farlo in locale o selezionare una delle seguenti opzioni per utilizzare Spark con Dataproc:

• Cluster Dataproc
• Dataproc Serverless

Per ulteriori informazioni sulla scelta tra un'opzione per cluster Dataproc e serverless, consulta la documentazione su Dataproc Serverless per Spark rispetto a Dataproc su Compute Engine .

Scarica il file JAR del connettore

Puoi trovare il codice sorgente del connettore Bigtable Spark con esempi nel repository GitHub del connettore Spark di Bigtable.

In base alla configurazione di Spark, puoi accedere al file JAR nel seguente modo:

• Se esegui PySpark in locale, devi scaricare il file JAR del connettore dal percorso Cloud Storage gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar.

  Sostituisci SCALA_VERSION con la versione Scala, impostata su 2.12 come unica versione supportata, e CONNECTOR_VERSION con la versione del connettore che vuoi utilizzare.

• Per l'opzione serverless o cluster Dataproc, utilizza il file JAR più recente come artefatto che può essere aggiunto nelle applicazioni Scala o Java Spark. Per saperne di più sull'utilizzo del file JAR come artefatto, consulta Gestire le dipendenze.

• Se invii il tuo job PySpark a Dataproc, utilizza il flag gcloud dataproc jobs submit pyspark --jars per impostare l'URI sulla posizione del file JAR in Cloud Storage, ad esempio gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar.

Aggiungi la configurazione Bigtable all'applicazione Spark

Nell'applicazione Spark, aggiungi le opzioni Spark che consentono di interagire con Bigtable.

Opzioni Spark supportate

Utilizza le opzioni Spark disponibili come parte del pacchetto com.google.cloud.spark.bigtable.

Nome opzione	Obbligatorio	Valore predefinito	Significato
spark.bigtable.project.id	Sì	N/A	Imposta l'ID progetto Bigtable.
spark.bigtable.instance.id	Sì	N/A	Imposta l'ID istanza Bigtable.
catalog	Sì	N/A	Imposta il formato JSON che specifica il formato di conversione compreso tra lo schema di tipo SQL del DataFrame e quello della tabella Bigtable. Per saperne di più, consulta Creare metadati delle tabelle in formato JSON.
spark.bigtable.app_profile.id	No	default	Imposta l'ID profilo dell'app Bigtable.
spark.bigtable.write.timestamp.milliseconds	No	Ora di sistema attuale	Imposta il timestamp in millisecondi da utilizzare durante la scrittura di un DataFrame in Bigtable. Tieni presente che, poiché tutte le righe nel DataFrame utilizzano lo stesso timestamp, le righe con la stessa colonna di chiave di riga nel DataFrame vengono mantenute come una singola versione in Bigtable poiché condividono lo stesso timestamp.
spark.bigtable.create.new.table	No	false	Imposta true per creare una nuova tabella prima di scrivere in Bigtable.
spark.bigtable.read.timerange.start.milliseconds o spark.bigtable.read.timerange.end.milliseconds	No	N/A	Imposta timestamp (in millisecondi dal momento dell'epoca) per filtrare le celle con una data di inizio e una data di fine specifiche, rispettivamente. È necessario specificare entrambi o nessuno di questi parametri.
spark.bigtable.push.down.row.key.filters	No	true	Imposta true per consentire l'applicazione di filtri a chiave di riga semplici sul lato server. Il filtro in base alle chiavi di riga composte viene implementato sul lato client. Per ulteriori informazioni, consulta Leggere una riga DataFrame specifica utilizzando un filtro.
spark.bigtable.read.rows.attempt.timeout.milliseconds	No	30 min	Imposta la durata del timeout per un tentativo di lettura di righe corrispondente a una partizione DataFrame nel client Bigtable per Java.
spark.bigtable.read.rows.total.timeout.milliseconds	No	12 ore	Imposta la durata del timeout totale per un tentativo di lettura di righe corrispondente a una partizione DataFrame nel client Bigtable per Java.
spark.bigtable.mutate.rows.attempt.timeout.milliseconds	No	1 mil	Imposta la durata del timeout per un tentativo di riga con modifica corrispondente a una partizione DataFrame nel client Bigtable per Java.
spark.bigtable.mutate.rows.total.timeout.milliseconds	No	10 min	Imposta la durata del timeout totale per un tentativo di modifica di righe corrispondente a una partizione DataFrame nel client Bigtable per Java.
spark.bigtable.batch.mutate.size	No	100	Impostato sul numero di mutazioni in ogni batch. Il valore massimo che puoi impostare è 100000.
spark.bigtable.enable.batch_mutate.flow_control	No	false	Imposta su true per abilitare il controllo del flusso per le mutazioni batch.

Crea metadati della tabella in formato JSON

Il formato della tabella DataFrames Spark SQL deve essere convertito in una tabella Bigtable utilizzando una stringa in formato JSON. Questo formato JSON di stringa rende il formato dei dati compatibile con Bigtable. Puoi passare il formato JSON nel codice dell'applicazione utilizzando l'opzione .option("catalog", catalog_json_string).

Considera ad esempio la seguente tabella DataFrame e la tabella Bigtable corrispondente.

In questo esempio, le colonne name e birthYear nel DataFrame vengono raggruppate nella famiglia di colonne info e rinominate rispettivamente in name e birth_year. Allo stesso modo, la colonna address viene archiviata nella famiglia di colonne location con lo stesso nome di colonna. La colonna id del DataFrame viene convertita nella chiave di riga di Bigtable.

Le chiavi di riga non hanno un nome di colonna dedicato in Bigtable e in questo esempio, id_rowkey viene utilizzato solo per indicare al connettore che questa è la colonna della chiave di riga. Puoi utilizzare qualsiasi nome per la colonna della chiave di riga e assicurarti di utilizzare lo stesso nome quando dichiari il campo "rowkey":"column_name" in formato JSON.

DataFrame					Tabella Bigtable = t1
Colonne					Chiave di riga	Famiglie di colonne
						informazioni		località
						Colonne		Colonne
id	name	birthYear	address		id_rowkey	name	birth_year	address

Il formato JSON per il catalogo è il seguente:

    """
    {
      "table": {"name": "t1"},
      "rowkey": "id_rowkey",
      "columns": {
        "id": {"cf": "rowkey", "col": "id_rowkey", "type": "string"},
        "name": {"cf": "info", "col": "name", "type": "string"},
        "birthYear": {"cf": "info", "col": "birth_year", "type": "long"},
        "address": {"cf": "location", "col": "address", "type": "string"}
      }
    }
    """

Le chiavi e i valori utilizzati nel formato JSON sono i seguenti:

Chiave catalogo	Valore catalogo	Formato JSON
tabella	Nome della tabella Bigtable.	"table":{"name":"t1"} Se la tabella non esiste, usa .option("spark.bigtable.create.new.table", "true") per crearne una.
riga	Nome della colonna che verrà utilizzata come chiave di riga di Bigtable. Assicurati che il nome della colonna DataFrame sia utilizzato come chiave di riga, ad esempio id_rowkey. Le chiavi composte sono accettate anche come chiavi di riga. Ad esempio: "rowkey":"name:address". Questo approccio potrebbe causare chiavi di riga che richiedono una scansione completa della tabella per tutte le richieste di lettura.	"rowkey":"id_rowkey",
colonne	Mappatura di ogni colonna DataFrame nella famiglia di colonne Bigtable ("cf") e nel nome della colonna ("col") corrispondenti. Il nome della colonna può essere diverso dal nome della colonna nella tabella DataFrame. I tipi di dati supportati includono string, long e binary.	"columns": {"id": {"cf": "rowkey", "col": "id_rowkey", "type": "string"}, "name": {"cf": "info", "col": "name", "type": "string"}, "birthYear": {"cf":"info", "col": "birth_year", "type": "long"}, "address": {"cf": "location", "col": "address", "type":"string"}}" In questo esempio, id_rowkey è la chiave di riga e info e location sono le famiglie di colonne.

Tipi di dati supportati

Il connettore supporta l'utilizzo dei tipi string, long e binary (array di byte) nel catalogo. Fino a quando non verrà aggiunto il supporto di altri tipi, ad esempio int e float, puoi convertirli manualmente in array di byte (BinaryType di Spark SQL) prima di utilizzare il connettore per scriverli su Bigtable.

Inoltre, puoi utilizzare Avro per serializzare tipi complessi come ArrayType. Per maggiori informazioni, vedi Serializzare i tipi di dati complessi utilizzando Apache Avro.

Scrivi in Bigtable

Utilizza la funzione .write() e le opzioni supportate per scrivere dati in Bigtable.

Java

Il seguente codice del repository GitHub utilizza Java e Maven per scrivere in Bigtable.

  String catalog = "{" +
        "\"table\":{\"name\":\"" + tableName + "\"," +
        "\"tableCoder\":\"PrimitiveType\"}," +
        "\"rowkey\":\"wordCol\"," +
        "\"columns\":{" +
        "\"word\":{\"cf\":\"rowkey\", \"col\":\"wordCol\", \"type\":\"string\"}," +
        "\"count\":{\"cf\":\"example_family\", \"col\":\"countCol\", \"type\":\"long\"}" +
        "}}".replaceAll("\\s+", "");

…

  private static void writeDataframeToBigtable(Dataset<Row> dataframe, String catalog,
        String createNewTable) {
      dataframe
          .write()
          .format("bigtable")
          .option("catalog", catalog)
          .option("spark.bigtable.project.id", projectId)
          .option("spark.bigtable.instance.id", instanceId)
          .option("spark.bigtable.create.new.table", createNewTable)
          .save();
    }

Python

Il seguente codice del repository GitHub utilizza Python per scrivere in Bigtable.

  catalog = ''.join(("""{
        "table":{"name":" """ + bigtable_table_name + """
        ", "tableCoder":"PrimitiveType"},
        "rowkey":"wordCol",
        "columns":{
          "word":{"cf":"rowkey", "col":"wordCol", "type":"string"},
          "count":{"cf":"example_family", "col":"countCol", "type":"long"}
        }
        }""").split())
  …

  input_data = spark.createDataFrame(data)
  print('Created the DataFrame:')
  input_data.show()

  input_data.write \
        .format('bigtable') \
        .options(catalog=catalog) \
        .option('spark.bigtable.project.id', bigtable_project_id) \
        .option('spark.bigtable.instance.id', bigtable_instance_id) \
        .option('spark.bigtable.create.new.table', create_new_table) \
        .save()
  print('DataFrame was written to Bigtable.')

  …

Lettura da Bigtable

Utilizza la funzione .read() per verificare se la tabella è stata importata correttamente in Bigtable.

Java

  …
  private static Dataset<Row> readDataframeFromBigtable(String catalog) {
      Dataset<Row> dataframe = spark
          .read()
          .format("bigtable")
          .option("catalog", catalog)
          .option("spark.bigtable.project.id", projectId)
          .option("spark.bigtable.instance.id", instanceId)
          .load();
      return dataframe;
    }

Python

  …

  records = spark.read \
        .format('bigtable') \
        .option('spark.bigtable.project.id', bigtable_project_id) \
        .option('spark.bigtable.instance.id', bigtable_instance_id) \
        .options(catalog=catalog) \
        .load()

  print('Reading the DataFrame from Bigtable:')
  records.show()

Compila il progetto

Genera il file JAR utilizzato per eseguire un job in un cluster Dataproc, Dataproc serverless o un'istanza Spark locale. Puoi compilare il file JAR localmente e quindi utilizzarlo per inviare un job. Il percorso del JAR compilato viene impostato come variabile di ambiente PATH_TO_COMPILED_JAR quando invii un job.

Questo passaggio non riguarda le applicazioni PySpark.

Gestisci le dipendenze

Il connettore Spark di Bigtable supporta i seguenti strumenti di gestione delle dipendenze:

• Maven
• sbt
• Gradle

Compila il file JAR

Maven

1. Aggiungi la dipendenza spark-bigtable al file pom.xml.

   <dependencies>
   <dependency>
     <groupId>com.google.cloud.spark.bigtable</groupId>
     <artifactId>spark-bigtable_SCALA_VERSION</artifactId>
     <version>0.1.0</version>
   </dependency>
   </dependencies>
   

2. Aggiungi il plug-in Maven Shade al tuo file pom.xml per creare un JAR uber:

   <plugins>
     <plugin>
       <groupId>org.apache.maven.plugins</groupId>
       <artifactId>maven-shade-plugin</artifactId>
       <version>3.2.4</version>
       <executions>
         <execution>
           <phase>package</phase>
           <goals>
             <goal>shade</goal>
           </goals>
         </execution>
       </executions>
     </plugin>
   </plugins>
   

3. Esegui il comando mvn clean install per generare un file JAR.

sbt

1. Aggiungi la dipendenza spark-bigtable al tuo file build.sbt:

   libraryDependencies += "com.google.cloud.spark.bigtable" % "spark-bigtable_SCALA_VERSION" % "0.1.0{""}}"
   

2. Aggiungi il plug-in sbt-assembly al file project/plugins.sbt o project/assembly.sbt per creare un file JAR Uber.

   addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")
   

3. Esegui il comando sbt clean assembly per generare il file JAR.

Gradle

1. Aggiungi la dipendenza spark-bigtable al file build.gradle.

   dependencies {
   implementation group: 'com.google.cloud.bigtable', name: 'spark-bigtable_SCALA_VERSION', version: '0.1.0'
   }
   

2. Aggiungi il plug-in Shadow nel file build.gradle per creare un file JAR Uber:

   plugins {
   id 'com.github.johnrengelman.shadow' version '8.1.1'
   id 'java'
   }
   

3. Consulta la documentazione del plug-in Shadow per ulteriori informazioni sulla configurazione e sulla compilazione JAR.

Nota: consulta la documentazione di Gradle per informazioni dettagliate sulla creazione di un file JAR Uber con Gradle.

invia un job

Invia un job Spark utilizzando Dataproc, Dataproc Serverless o un'istanza Spark locale per avviare l'applicazione.

Imposta ambiente di runtime

Imposta le seguenti variabili di ambiente.

      #Google Cloud
      export BIGTABLE_SPARK_PROJECT_ID=PROJECT_ID
      export BIGTABLE_SPARK_INSTANCE_ID=INSTANCE_ID
      export BIGTABLE_SPARK_TABLE_NAME=TABLE_NAME
      export BIGTABLE_SPARK_DATAPROC_CLUSTER=DATAPROC_CLUSTER
      export BIGTABLE_SPARK_DATAPROC_REGION=DATAPROC_REGION
      export BIGTABLE_SPARK_DATAPROC_ZONE=DATAPROC_ZONE

      #Dataproc Serverless
      export BIGTABLE_SPARK_SUBNET=SUBNET
      export BIGTABLE_SPARK_GCS_BUCKET_NAME=GCS_BUCKET_NAME

      #Scala/Java
      export PATH_TO_COMPILED_JAR=PATH_TO_COMPILED_JAR

      #PySpark
      export GCS_PATH_TO_CONNECTOR_JAR=GCS_PATH_TO_CONNECTOR_JAR
      export PATH_TO_PYTHON_FILE=PATH_TO_PYTHON_FILE
      export LOCAL_PATH_TO_CONNECTOR_JAR=LOCAL_PATH_TO_CONNECTOR_JAR

Sostituisci quanto segue:

• PROJECT_ID: l'identificatore permanente per il progetto Bigtable.
• INSTANCE_ID: l'identificatore permanente dell'istanza Bigtable.
• TABLE_NAME: l'identificatore permanente della tabella.
• DATAPROC_CLUSTER: l'identificatore permanente per il cluster Dataproc.
• DATAPROC_REGION: la regione Dataproc che contiene uno dei cluster nella tua istanza Dataproc, ad esempio northamerica-northeast2.
• DATAPROC_ZONE: la zona in cui viene eseguito il cluster Dataproc.
• SUBNET: il percorso completo della risorsa della subnet.
• GCS_BUCKET_NAME: il bucket Cloud Storage in cui caricare le dipendenze dei carichi di lavoro Spark.
• PATH_TO_COMPILED_JAR: il percorso completo o relativo del file JAR compilato, ad esempio /path/to/project/root/target/<compiled_JAR_name> per Maven.
• GCS_PATH_TO_CONNECTOR_JAR: il bucket Cloud Storage gs://spark-lib/bigtable in cui si trova il file spark-bigtable_SCALA_VERSION_CONNECTOR_VERSION.jar.
• PATH_TO_PYTHON_FILE: per le applicazioni PySpark, il percorso del file Python che verrà utilizzato per scrivere e leggere i dati da Bigtable.
• LOCAL_PATH_TO_CONNECTOR_JAR: per le applicazioni PySpark, percorso del file JAR del connettore Bigtable Spark scaricato.

Invia un job Spark

Per le istanze Dataproc o la tua configurazione Spark locale, esegui un job Spark per caricare i dati su Bigtable.

Nota: per gli esempi di questa sezione, gli argomenti della riga di comando impostati e utilizzati servono per inviare gli esempi dal repository GitHub del connettore Spark di Bigtable. Un argomento di esempio per l'ID progetto è impostato in Java o Scala come $BIGTABLE_SPARK_PROJECT_ID, mentre in PySpark come
--bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID. Se utilizzi un tuo progetto, imposta gli argomenti della riga di comando di conseguenza.

Cluster Dataproc

Utilizza il file JAR compilato e crea un job del cluster Dataproc che legge e scrive i dati da e in Bigtable.

1. Creare un cluster Dataproc. L'esempio seguente mostra un comando per creare un cluster Dataproc v2.0 con Debian 10, due nodi worker e configurazioni predefinite.

   gcloud dataproc clusters create \
     $BIGTABLE_SPARK_DATAPROC_CLUSTER --region $BIGTABLE_SPARK_DATAPROC_REGION \
     --zone $BIGTABLE_SPARK_DATAPROC_ZONE \
     --master-machine-type n2-standard-4 --master-boot-disk-size 500 \
     --num-workers 2 --worker-machine-type n2-standard-4 --worker-boot-disk-size 500 \
     --image-version 2.0-debian10 --project $BIGTABLE_SPARK_PROJECT_ID
   

   Nota: assicurati di includere un framework di logging compatibile con l'API di logging SLF4J. Per le applicazioni PySpark, utilizza l'opzione
   --packages per specificare il pacchetto di logging, ad esempio reload4j, quando invii il job Spark. Per le applicazioni Scala e Java, includi il pacchetto di logging nello strumento di gestione delle dipendenze. Se non includi questo pacchetto, si verifica l'errore java.lang.NoClassDefFoundError: org/slf4j/impl/StaticLoggerBinder.

2. Invia un job.

   Scala/Java

   L'esempio seguente mostra la classe spark.bigtable.example.WordCount che include la logica per creare una tabella di test in DataFrame, scrivere la tabella su Bigtable e quindi contare il numero di parole nella tabella.

       gcloud dataproc jobs submit spark \
       --cluster=$BIGTABLE_SPARK_DATAPROC_CLUSTER \
       --region=$BIGTABLE_SPARK_DATAPROC_REGION \
       --class=spark.bigtable.example.WordCount \
       --jar=$PATH_TO_COMPILED_JAR \
       -- \
       $BIGTABLE_SPARK_PROJECT_ID \
       $BIGTABLE_SPARK_INSTANCE_ID \
       $BIGTABLE_SPARK_TABLE_NAME \
   

   PySpark

       gcloud dataproc jobs submit pyspark \
       --cluster=$BIGTABLE_SPARK_DATAPROC_CLUSTER \
       --region=$BIGTABLE_SPARK_DATAPROC_REGION \
       --jars=$GCS_PATH_TO_CONNECTOR_JAR \
       --properties='spark.jars.packages=org.slf4j:slf4j-reload4j:1.7.36' \
       $PATH_TO_PYTHON_FILE \
       -- \
       --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
       --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
       --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME \
   

Dataproc Serverless

Utilizza il file JAR compilato e crea un job Dataproc che legge e scrive dati da e in Bigtable con un'istanza serverless Dataproc.

Scala/Java

  gcloud dataproc batches submit spark \
  --region=$BIGTABLE_SPARK_DATAPROC_REGION \
  --subnet=$BIGTABLE_SPARK_SUBNET --version=1.1 \
  --deps-bucket=gs://$BIGTABLE_SPARK_GCS_BUCKET_NAME --jar=$PATH_TO_COMPILED_JAR \
  --  \
  $BIGTABLE_SPARK_PROJECT_ID \
  $BIGTABLE_SPARK_INSTANCE_ID \
  $BIGTABLE_SPARK_TABLE_NAME

PySpark

  gcloud dataproc batches submit pyspark $PATH_TO_PYTHON_FILE \
  --region=$BIGTABLE_SPARK_DATAPROC_REGION \
  --subnet=$BIGTABLE_SPARK_SUBNET --version=1.1 \
  --deps-bucket=gs://$BIGTABLE_SPARK_GCS_BUCKET_NAME \
  --jars=$GCS_PATH_TO_CONNECTOR_JAR \
  --properties='spark.jars.packages=org.slf4j:slf4j-reload4j:1.7.36' \
  -- \
  --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
  --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
  --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME

Nota: se si verifica un errore Specify a network which contains subnetworks to enable 'internal_ip_only' durante l'invio di un job a Dataproc Serverless, prova a creare una nuova subnet. Per saperne di più sulla creazione di una nuova subnet, consulta Creare e gestire le reti VPC.

Spark locale

Utilizzare il file JAR scaricato e creare un job Spark che legge e scrive i dati da e in Bigtable con un'istanza Spark locale. Puoi anche utilizzare l'emulatore Bigtable per inviare il job Spark.

Usa l'emulatore Bigtable

Se decidi di utilizzare l'emulatore Bigtable, segui questi passaggi:

1. Esegui questo comando per avviare l'emulatore:

   gcloud beta emulators bigtable start
   

   Per impostazione predefinita, l'emulatore sceglie localhost:8086.

2. Imposta la variabile di ambiente BIGTABLE_EMULATOR_HOST:

   export BIGTABLE_EMULATOR_HOST=localhost:8086
   

3. Invia il job Spark.

Per ulteriori informazioni sull'utilizzo dell'emulatore Bigtable, consulta Eseguire test con l'emulatore.

Invia un job Spark

Usa il comando spark-submit per inviare un job Spark a prescindere dal fatto che tu stia utilizzando un emulatore Bigtable locale.

Scala/Java

  spark-submit $PATH_TO_COMPILED_JAR \
  $BIGTABLE_SPARK_PROJECT_ID \
  $BIGTABLE_SPARK_INSTANCE_ID \
  $BIGTABLE_SPARK_TABLE_NAME

PySpark

  spark-submit \
  --jars=$LOCAL_PATH_TO_CONNECTOR_JAR \
  --packages=org.slf4j:slf4j-reload4j:1.7.36 \
  $PATH_TO_PYTHON_FILE \
  --bigtableProjectId=$BIGTABLE_SPARK_PROJECT_ID \
  --bigtableInstanceId=$BIGTABLE_SPARK_INSTANCE_ID \
  --bigtableTableName=$BIGTABLE_SPARK_TABLE_NAME

Verificare i dati della tabella

Esegui questo comando dell'interfaccia a riga di comando cbt per verificare che i dati siano scritti in Bigtable. cbt CLI è un componente di Google Cloud CLI. Per maggiori informazioni, consulta la panoramica dell'interfaccia a riga di comando di cbt.

    cbt -project=$BIGTABLE_SPARK_PROJECT_ID -instance=$BIGTABLE_SPARK_INSTANCE_ID \
    read $BIGTABLE_SPARK_TABLE_NAME

Soluzioni aggiuntive

Utilizzare il connettore Spark di Bigtable per soluzioni specifiche, come serializzare tipi di Spark SQL complessi, leggere righe specifiche e generare metriche lato client.

Leggere una riga DataFrame specifica utilizzando un filtro

Quando utilizzi DataFrames per leggere da Bigtable, puoi specificare un filtro per leggere solo righe specifiche. Semplici filtri come ==, <= e startsWith nella colonna della chiave di riga vengono applicati sul lato server per evitare una scansione completa della tabella. I filtri su chiavi di riga composte o filtri complessi come il filtro LIKE nella colonna della chiave di riga vengono applicati sul lato client.

Se stai leggendo tabelle di grandi dimensioni, consigliamo di utilizzare semplici filtri con chiave di riga per evitare di eseguire una scansione completa della tabella. La seguente istruzione di esempio mostra come leggere utilizzando un filtro semplice. Assicurati di utilizzare nel filtro Spark il nome della colonna DataFrame convertita nella chiave di riga:

    dataframe.filter("id == 'some_id'").show()
  

Quando applichi un filtro, utilizza il nome della colonna DataFrame anziché il nome della colonna della tabella Bigtable.

Serializza tipi di dati complessi utilizzando Apache Avro

Il connettore Spark di Bigtable fornisce supporto per l'utilizzo di Apache Avro per serializzare tipi SQL Spark complessi, come ArrayType, MapType o StructType. Apache Avro fornisce la serializzazione dei dati per registrare i dati, che vengono comunemente utilizzati per l'elaborazione e l'archiviazione di strutture di dati complesse.

Utilizza una sintassi come "avro":"avroSchema" per specificare che una colonna in Bigtable deve essere codificata utilizzando Avro. Puoi quindi utilizzare .option("avroSchema", avroSchemaString) durante la lettura o la scrittura su Bigtable per specificare lo schema Avro corrispondente a quella colonna in formato stringa. Puoi utilizzare nomi di opzioni diversi, ad esempio "anotherAvroSchema" per colonne diverse e trasmettere schemi Avro per più colonne.

def catalogWithAvroColumn = s"""{
                    |"table":{"name":"ExampleAvroTable"},
                    |"rowkey":"key",
                    |"columns":{
                    |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
                    |"col1":{"cf":"cf1", "col":"col1", "avro":"avroSchema"}
                    |}
                    |}""".stripMargin

Utilizza le metriche lato client

Poiché il connettore Spark di Bigtable si basa sul client Bigtable per Java, le metriche lato client sono abilitate all'interno del connettore per impostazione predefinita. Per ulteriori informazioni su come accedere e interpretare queste metriche, puoi consultare la documentazione relativa alle metriche lato client.

Utilizzare il client Bigtable per Java con funzioni RDD di basso livello

Poiché il connettore Spark di Bigtable si basa sul client Bigtable per Java, puoi utilizzare direttamente il client nelle applicazioni Spark ed eseguire richieste di lettura o scrittura distribuite all'interno delle funzioni RDD di basso livello, come mapPartitions e foreachPartition.

Per utilizzare il client Bigtable per le classi Java, aggiungi il prefisso com.google.cloud.spark.bigtable.repackaged ai nomi dei pacchetti. Ad esempio, invece di utilizzare il nome della classe com.google.cloud.bigtable.data.v2.BigtableDataClient, usa com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.BigtableDataClient.

Per saperne di più sul client Bigtable per Java, consulta la pagina Client Bigtable per Java.

Passaggi successivi

• Scopri come ottimizzare il tuo job Spark in Dataproc.
• Utilizza le classi del client Bigtable per Java con il connettore Bigtable Spark.