Bigtable Spark-Connector verwenden

Mit dem Bigtable Spark-Connector können Sie Daten von und nach Bigtable lesen und schreiben. Sie können Daten mit Spark SQL und DataFrames aus Ihrer Spark-Anwendung lesen. Die folgenden Bigtable-Vorgänge werden über den Bigtable Spark-Connector unterstützt:

  • Daten schreiben
  • Daten lesen
  • Neue Tabelle erstellen

In diesem Dokument erfahren Sie, wie Sie eine Spark SQL DataFrames-Tabelle in eine Bigtable-Tabelle konvertieren und dann eine JAR-Datei kompilieren und erstellen, um einen Spark-Job einzureichen.

Supportstatus für Spark und Scala

Der Bigtable Spark-Connector unterstützt nur die Scala-Version 2.12 und die folgenden Spark-Versionen:

Der Bigtable Spark-Connector unterstützt die folgenden Dataproc-Versionen:

Kosten berechnen

Wenn Sie eine der folgenden abrechenbaren Google Cloud-Komponenten verwenden, werden Ihnen die verwendeten Ressourcen in Rechnung gestellt:

  • Bigtable (die Nutzung des Bigtable-Emulators ist kostenlos)
  • Dataproc
  • Cloud Storage

Die Dataproc-Preise gelten für die Verwendung von Dataproc in Compute Engine-Clustern. Die Preise für Dataproc Serverless gelten für Arbeitslasten und Sitzungen, die in Dataproc Serverless für Spark ausgeführt werden.

Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen.

Hinweise

Führen Sie die folgenden Voraussetzungen aus, bevor Sie den Bigtable Spark-Connector verwenden.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Verwenden des Bigtable Spark-Connectors benötigen:

  • Bigtable-Administrator (roles/bigtable.admin)(optional): Ermöglicht das Lesen oder Schreiben von Daten und das Erstellen einer neuen Tabelle.
  • Bigtable-Nutzer (roles/bigtable.user): Sie können damit Daten lesen oder schreiben, aber keine neue Tabelle erstellen.

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Wenn Sie Dataproc oder Cloud Storage verwenden, sind möglicherweise zusätzliche Berechtigungen erforderlich. Weitere Informationen finden Sie unter Dataproc-Berechtigungen und Cloud Storage-Berechtigungen.

Spark einrichten

Sie müssen nicht nur eine Bigtable-Instanz erstellen, sondern auch Ihre Spark-Instanz einrichten. Sie können dies lokal tun oder eine der folgenden Optionen auswählen, um Spark mit Dataproc zu verwenden:

  • Dataproc-Cluster
  • Dataproc Serverless

Weitere Informationen zur Auswahl zwischen einem Dataproc-Cluster oder einer serverlosen Option finden Sie in der Dokumentation Dataproc Serverless für Spark im Vergleich zu Dataproc in der Compute Engine .

JAR-Datei des Connectors herunterladen

Den Quellcode des Bigtable Spark-Connectors mit Beispielen finden Sie im GitHub-Repository des Bigtable Spark-Connectors.

Je nach Spark-Konfiguration können Sie auf die JAR-Datei so zugreifen:

  • Wenn Sie PySpark lokal ausführen, sollten Sie die JAR-Datei des Connectors von gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar aus Cloud Storage herunterladen.

    Ersetzen Sie SCALA_VERSION durch die Scala-Version, die als einzige unterstützte Version auf 2.12 festgelegt ist, und CONNECTOR_VERSION durch die zu verwendende Connector-Version.

  • Verwenden Sie für einen Dataproc-Cluster oder die serverlose Option die neueste JAR-Datei als Artefakt, das Ihren Scala- oder Java-Spark-Anwendungen hinzugefügt werden kann. Weitere Informationen zur Verwendung der JAR-Datei als Artefakt finden Sie unter Abhängigkeiten verwalten.

  • Wenn Sie Ihren PySpark-Job an Dataproc senden, verwenden Sie das Flag gcloud dataproc jobs submit pyspark --jars, um den URI auf den Speicherort der JAR-Datei in Cloud Storage festzulegen, z. B. gs://spark-lib/bigtable/spark-bigtable_SCALA_VERSION-CONNECTOR_VERSION.jar.

Rechentyp ermitteln

Für Lesejobs können Sie serverloses Computing mit Data Boost (Vorabversion) verwenden, um Auswirkungen auf die Cluster zu vermeiden, die Ihre Anwendungen bereitstellen. Ihre Spark-Anwendung muss Version 1.1.0 oder höher des Spark-Connectors verwenden, um Data Boost nutzen zu können.

Wenn Sie Data Boost verwenden möchten, müssen Sie ein Data Boost-Anwendungsprofil erstellen und dann die App-Profil-ID für die spark.bigtable.app_profile.id Spark-Option angeben, wenn Sie Ihrer Spark-Anwendung die Bigtable-Konfiguration hinzufügen. Wenn Sie bereits ein Anwendungsprofil für Ihre Spark-Lesejobs erstellt haben und es weiter verwenden möchten, ohne den Anwendungscode zu ändern, können Sie das Anwendungsprofil in ein Data Boost-Anwendungsprofil umwandeln. Weitere Informationen finden Sie unter App-Profil konvertieren.

Weitere Informationen finden Sie in der Übersicht über Bigtable Data Boost.

Bei Jobs mit Lese- und Schreibvorgängen können Sie die Clusterknoten Ihrer Instanz für die Datenverarbeitung verwenden, indem Sie in Ihrer Anfrage ein Standardanwendungsprofil angeben.

Zu verwendendes App-Profil ermitteln oder erstellen

Wenn Sie keine App-Profil-ID angeben, verwendet der Connector das Standard-App-Profil.

Wir empfehlen, für jede Anwendung, die Sie ausführen, ein eigenes App-Profil zu verwenden, einschließlich Ihrer Spark-Anwendung. Weitere Informationen zu den Typen und Einstellungen von Anwendungsprofilen finden Sie in der Übersicht über Anwendungsprofile. Eine Anleitung finden Sie unter Anwendungsprofile erstellen und konfigurieren.

Bigtable-Konfiguration zur Spark-Anwendung hinzufügen

Fügen Sie in Ihrer Spark-Anwendung die Spark-Optionen hinzu, mit denen Sie mit Bigtable interagieren können.

Unterstützte Spark-Optionen

Verwenden Sie die Spark-Optionen, die im Paket com.google.cloud.spark.bigtable verfügbar sind.

Option Erforderlich Standardwert Bedeutung
spark.bigtable.project.id Ja – Legen Sie die Bigtable-Projekt-ID fest.
spark.bigtable.instance.id Ja – Legen Sie die Bigtable-Instanz-ID fest.
catalog Ja – Legen Sie das JSON-Format fest, das das Konvertierungsformat zwischen dem SQL-ähnlichen Schema des DataFrames und dem Schema der Bigtable-Tabelle angibt.

Weitere Informationen finden Sie unter Tabellenmetadaten im JSON-Format erstellen.
spark.bigtable.app_profile.id Nein default Legen Sie die Bigtable-Anwendungsprofil-ID fest.
spark.bigtable.write.timestamp.milliseconds Nein Aktuelle Systemzeit Legen Sie den Zeitstempel in Millisekunden fest, der beim Schreiben eines DataFrames in Bigtable verwendet werden soll.

Da alle Zeilen im DataFrame denselben Zeitstempel verwenden, werden Zeilen mit derselben Zeilenschlüsselspalte im DataFrame als einzelne Version in Bigtable beibehalten, da sie denselben Zeitstempel haben.
spark.bigtable.create.new.table Nein false Legen Sie true fest, um eine neue Tabelle zu erstellen, bevor Sie in Bigtable schreiben.
spark.bigtable.read.timerange.start.milliseconds oder spark.bigtable.read.timerange.end.milliseconds Nein – Legen Sie Zeitstempel (in Millisekunden seit der Epoche) fest, um Zellen mit einem bestimmten Start- bzw. Enddatum zu filtern.
spark.bigtable.push.down.row.key.filters Nein true Legen Sie den Wert auf true fest, um ein einfaches Zeilenschlüssel-Filtern auf der Serverseite zu ermöglichen. Das Filtern nach zusammengesetzten Zeilenschlüsseln wird auf der Clientseite implementiert.

Weitere Informationen finden Sie unter Bestimmte DataFrame-Zeile mit einem Filter lesen.
spark.bigtable.read.rows.attempt.timeout.milliseconds Nein 30 m Legen Sie die Zeitüberschreitung für einen Lesezeilenversuch fest, der einer DataFrame-Partition im Bigtable-Client für Java entspricht.
spark.bigtable.read.rows.total.timeout.milliseconds Nein 12 Std. Legen Sie die Gesamt-Zeitüberschreitung für einen Lesezeilenversuch fest, der einer DataFrame-Partition im Bigtable-Client für Java entspricht.
spark.bigtable.mutate.rows.attempt.timeout.milliseconds Nein 1m Legen Sie die Zeitüberschreitung für einen Versuch zum Ändern von Zeilen fest, der einer DataFrame-Partition im Bigtable-Client für Java entspricht.
spark.bigtable.mutate.rows.total.timeout.milliseconds Nein 10 Min. Legen Sie die Gesamt-Zeitüberschreitung für einen Versuch zum Ändern von Zeilen fest, der einer DataFrame-Partition im Bigtable-Client für Java entspricht.
spark.bigtable.batch.mutate.size Nein 100 Legen Sie die Anzahl der Mutationen in jedem Batch fest. Der Höchstwert ist 100000.
spark.bigtable.enable.batch_mutate.flow_control Nein false Legen Sie true fest, um die Ablaufsteuerung für Batchmutationen zu aktivieren.

Tabellenmetadaten im JSON-Format erstellen

Das Spark SQL DataFrames-Tabellenformat muss mit einem String im JSON-Format in eine Bigtable-Tabelle konvertiert werden. Dieses JSON-Stringformat macht das Datenformat mit Bigtable kompatibel. Sie können das JSON-Format mit der Option .option("catalog", catalog_json_string) in Ihrem Anwendungscode übergeben.

Betrachten Sie als Beispiel die folgende DataFrame-Tabelle und die entsprechende Bigtable-Tabelle.

In diesem Beispiel werden die Spalten name und birthYear im DataFrame in der Spaltenfamilie info gruppiert und in name bzw. birth_year umbenannt. Die Spalte address wird ebenfalls unter der Spaltenfamilie location mit demselben Spaltennamen gespeichert. Die Spalte id aus dem DataFrame wird in den Bigtable-Zeilenschlüssel umgewandelt.

Die Zeilenschlüssel haben in Bigtable keinen eigenen Spaltennamen. In diesem Beispiel wird id_rowkey nur verwendet, um dem Connector anzugeben, dass dies die Spalte für den Zeilenschlüssel ist. Sie können einen beliebigen Namen für die Zeile mit dem Schlüssel verwenden. Achten Sie darauf, dass Sie denselben Namen verwenden, wenn Sie das Feld "rowkey":"column_name" im JSON-Format deklarieren.

DataFrame Bigtable-Tabelle = t1
Spalten Zeilenschlüssel Spaltenfamilien
Info Standort
Spalten Spalten
id name birthYear Adresse id_rowkey name birth_year Adresse

Das JSON-Format für den Katalog sieht so aus:

    """
    {
      "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"}
      }
    }
    """

Die im JSON-Format verwendeten Schlüssel und Werte sind:

Katalogschlüssel Katalogwert JSON-Format
table Name der Bigtable-Tabelle. "table":{"name":"t1"}

Wenn die Tabelle nicht vorhanden ist, erstellen Sie sie mit .option("spark.bigtable.create.new.table", "true").
rowkey Name der Spalte, die als Bigtable-Zeilenschlüssel verwendet wird. Der Spaltenname der DataFrame-Spalte muss als Zeilenschlüssel verwendet werden, z. B. id_rowkey.

Kompositschlüssel werden auch als Zeilenschlüssel akzeptiert. Beispiel: "rowkey":"name:address" Dieser Ansatz kann zu Zeilenschlüsseln führen, für die für alle Leseanfragen ein vollständiger Tabellenscan erforderlich ist.		 "rowkey":"id_rowkey",
Spalten Zuordnung jeder DataFrame-Spalte zur entsprechenden Bigtable-Spaltenfamilie ("cf") und zum Spaltennamen ("col"). Der Spaltenname kann sich vom Spaltennamen in der DataFrame-Tabelle unterscheiden. Zu den unterstützten Datentypen gehören string, long und 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 diesem Beispiel ist id_rowkey der Zeilenschlüssel und info und location sind die Spaltenfamilien.

Unterstützte Datentypen

Der Connector unterstützt die Verwendung der Typen string, long und binary (Byte-Array) im Katalog. Bis die Unterstützung für andere Typen wie int und float hinzugefügt wird, können Sie solche Datentypen manuell in Byte-Arrays (BinaryType in Spark SQL) konvertieren, bevor Sie sie über den Connector in Bigtable schreiben.

Außerdem können Sie Avro zum Serialisieren komplexer Typen wie ArrayType verwenden. Weitere Informationen finden Sie unter Komplexe Datentypen mit Apache Avro serialisieren.

In Bigtable schreiben

Verwenden Sie die Funktion .write() und die unterstützten Optionen, um Ihre Daten in Bigtable zu schreiben.

Java

Im folgenden Code aus dem GitHub-Repository wird mit Java und Maven in Bigtable geschrieben.

  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

Im folgenden Code aus dem GitHub-Repository wird Python verwendet, um Daten in Bigtable zu schreiben.

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

  …

Aus Bigtable lesen

Mit der Funktion .read() können Sie prüfen, ob die Tabelle erfolgreich in Bigtable importiert wurde.

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()

Projekt kompilieren

Erstellen Sie die JAR-Datei, mit der ein Job entweder in einem Dataproc-Cluster, in Dataproc Serverless oder in einer lokalen Spark-Instanz ausgeführt wird. Sie können die JAR-Datei lokal kompilieren und dann einen Job damit einreichen. Der Pfad zum kompilierten JAR wird beim Einreichen eines Jobs als Umgebungsvariable PATH_TO_COMPILED_JAR festgelegt.

Dieser Schritt gilt nicht für PySpark-Anwendungen.

Abhängigkeiten verwalten

Der Bigtable Spark-Connector unterstützt die folgenden Tools zur Abhängigkeitsverwaltung:

JAR-Datei kompilieren

Maven

  1. Fügen Sie der Datei pom.xml die Abhängigkeit spark-bigtable hinzu.

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

  2. Fügen Sie der pom.xml-Datei das Maven-Shade-Plug-in hinzu, um eine Uber-JAR-Datei zu erstellen:

    <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. Führen Sie den Befehl mvn clean install aus, um eine JAR-Datei zu generieren.

sbt

  1. Fügen Sie der Datei build.sbt die Abhängigkeit spark-bigtable hinzu:

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

  2. Fügen Sie Ihrer project/plugins.sbt- oder project/assembly.sbt-Datei das sbt-assembly-Plug-in hinzu, um eine Uber-JAR-Datei zu erstellen.

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

  3. Führen Sie den Befehl sbt clean assembly aus, um die JAR-Datei zu generieren.

Gradle

  1. Fügen Sie der Datei build.gradle die Abhängigkeit spark-bigtable hinzu.

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

  2. Fügen Sie das Shadow-Plug-in in die Datei build.gradle ein, um eine Uber-JAR-Datei zu erstellen:

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

  3. Weitere Informationen zur Konfiguration und JAR-Kompilierung finden Sie in der Dokumentation zum Shadow-Plug-in.

Job senden

Reichen Sie einen Spark-Job mit Dataproc, Dataproc Serverless oder einer lokalen Spark-Instanz ein, um Ihre Anwendung zu starten.

Laufzeitumgebung festlegen

Legen Sie die folgenden Umgebungsvariablen fest:

      #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

Ersetzen Sie Folgendes:

  • PROJECT_ID: Die permanente Kennzeichnung des Bigtable-Projekts.
  • INSTANCE_ID: Die permanente Kennzeichnung der Bigtable-Instanz.
  • TABLE_NAME: Die permanente Kennzeichnung der Tabelle.
  • DATAPROC_CLUSTER: Die permanente Kennzeichnung des Dataproc-Clusters.
  • DATAPROC_REGION: Die Dataproc-Region, die einen der Cluster in Ihrer Dataproc-Instanz enthält, z. B. northamerica-northeast2.
  • DATAPROC_ZONE: Die Zone, in der der Dataproc-Cluster ausgeführt wird.
  • SUBNET: Der vollständige Ressourcenpfad des Subnetzes.
  • GCS_BUCKET_NAME: Der Cloud Storage-Bucket, in den Spark-Workload-Abhängigkeiten hochgeladen werden sollen.
  • PATH_TO_COMPILED_JAR: Der vollständige oder relative Pfad zur kompilierten JAR-Datei, z. B. /path/to/project/root/target/<compiled_JAR_name> für Maven.
  • GCS_PATH_TO_CONNECTOR_JAR: Der Cloud Storage-Bucket gs://spark-lib/bigtable, in dem sich die Datei spark-bigtable_SCALA_VERSION_CONNECTOR_VERSION.jar befindet.
  • PATH_TO_PYTHON_FILE: Bei PySpark-Anwendungen der Pfad zur Python-Datei, mit der Daten in Bigtable geschrieben und aus Bigtable gelesen werden.
  • LOCAL_PATH_TO_CONNECTOR_JAR: Bei PySpark-Anwendungen: Pfad zur heruntergeladenen JAR-Datei des Bigtable Spark-Connectors.

Spark-Job senden

Führen Sie für Dataproc-Instanzen oder Ihre lokale Spark-Umgebung einen Spark-Job aus, um Daten in Bigtable hochzuladen.

Dataproc-Cluster

Verwenden Sie die kompilierte JAR-Datei und erstellen Sie einen Dataproc-Clusterjob, der Daten in Bigtable liest und schreibt.

  1. Erstellen Sie einen Dataproc-Cluster. Im folgenden Beispiel wird ein Beispielbefehl zum Erstellen eines Dataproc v2.0-Clusters mit Debian 10, zwei Worker-Knoten und Standardkonfigurationen gezeigt.

    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

  2. Senden Sie einen Job.

    Scala/Java

    Das folgende Beispiel zeigt die Klasse spark.bigtable.example.WordCount mit der Logik zum Erstellen einer Testtabelle in DataFrame, zum Schreiben der Tabelle in Bigtable und zum Zählen der Anzahl der Wörter in der Tabelle.

        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

Verwenden Sie die kompilierte JAR-Datei und erstellen Sie einen Dataproc-Job, der Daten mit einer Dataproc Serverless-Instanz in Bigtable liest und schreibt.

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

Lokaler Spark

Verwenden Sie die heruntergeladene JAR-Datei und erstellen Sie einen Spark-Job, der Daten mit einer lokalen Spark-Instanz von und nach Bigtable liest und schreibt. Sie können den Spark-Job auch über den Bigtable-Emulator einreichen.

Bigtable-Emulator verwenden

Wenn Sie den Bigtable-Emulator verwenden möchten, gehen Sie so vor:

  1. Verwenden Sie den folgenden Befehl, um den Emulator zu starten:

    gcloud beta emulators bigtable start

    Standardmäßig wählt der Emulator localhost:8086 aus.

  2. Legen Sie die Umgebungsvariable BIGTABLE_EMULATOR_HOST fest:

    export BIGTABLE_EMULATOR_HOST=localhost:8086

  3. Senden Sie den Spark-Job.

Weitere Informationen zur Verwendung des Bigtable-Emulators finden Sie unter Mit dem Emulator testen.

Spark-Job senden

Verwenden Sie den Befehl spark-submit, um einen Spark-Job einzureichen, unabhängig davon, ob Sie einen lokalen Bigtable-Emulator verwenden.

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

Tabellendaten prüfen

Führen Sie den folgenden cbt CLI-Befehl aus, um zu prüfen, ob die Daten in Bigtable geschrieben wurden. Die cbt CLI ist eine Komponente der Google Cloud CLI. Weitere Informationen finden Sie in der Übersicht über die cbt-CLI.

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

Zusätzliche Lösungen

Verwenden Sie den Bigtable Spark-Connector für bestimmte Lösungen, z. B. zum Serialisieren komplexer Spark SQL-Typen, zum Lesen bestimmter Zeilen und zum Generieren clientseitiger Messwerte.

Bestimmte DataFrame-Zeile mit einem Filter lesen

Wenn Sie DataFrames zum Lesen aus Bigtable verwenden, können Sie einen Filter angeben, um nur bestimmte Zeilen zu lesen. Einfache Filter wie ==, <= und startsWith werden serverseitig auf die Zeilenschlüsselspalte angewendet, um einen vollständigen Tabellenscan zu vermeiden. Filter für zusammengesetzte Zeilenschlüssel oder komplexe Filter wie der LIKE-Filter für die Zeilenschlüsselspalte werden auf der Clientseite angewendet.

Wenn Sie große Tabellen lesen, empfehlen wir die Verwendung einfacher Zeilenschlüsselfilter, um einen vollständigen Tabellenscan zu vermeiden. In der folgenden Beispielanweisung wird gezeigt, wie mit einem einfachen Filter gelesen wird. Achten Sie darauf, in Ihrem Spark-Filter den Namen der DataFrame-Spalte zu verwenden, die in den Zeilenschlüssel umgewandelt wird:

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

Verwenden Sie beim Anwenden eines Filters den DataFrame-Spaltennamen anstelle des Bigtable-Tabellenspaltennamens.

Komplexe Datentypen mit Apache Avro serialisieren

Der Bigtable Spark-Connector unterstützt die Verwendung von Apache Avro zum Serialisieren komplexer Spark SQL-Typen wie ArrayType, MapType oder StructType. Apache Avro bietet eine Datenserialisierung für Datensätze, die häufig zum Verarbeiten und Speichern komplexer Datenstrukturen verwendet wird.

Verwenden Sie eine Syntax wie "avro":"avroSchema", um anzugeben, dass eine Spalte in Bigtable mit Avro codiert werden soll. Sie können dann .option("avroSchema", avroSchemaString) verwenden, um beim Lesen von oder Schreiben in Bigtable das Avro-Schema anzugeben, das dieser Spalte im Stringformat entspricht. Sie können unterschiedliche Optionennamen verwenden, z. B. "anotherAvroSchema" für verschiedene Spalten, und Avro-Schemas für mehrere Spalten übergeben.

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

Clientseitige Messwerte verwenden

Da der Bigtable Spark-Connector auf dem Bigtable-Client für Java basiert, sind clientseitige Messwerte im Connector standardmäßig aktiviert. Weitere Informationen zum Zugriff auf und zur Interpretation dieser Messwerte finden Sie in der Dokumentation zu clientseitigen Messwerten.

Bigtable-Client für Java mit Low-Level-RDD-Funktionen verwenden

Da der Bigtable Spark-Connector auf dem Bigtable-Client für Java basiert, können Sie den Client direkt in Ihren Spark-Anwendungen verwenden und verteilte Lese- oder Schreibanfragen innerhalb der Low-Level-RDD-Funktionen wie mapPartitions und foreachPartition ausführen.

Wenn Sie den Bigtable-Client für Java-Klassen verwenden möchten, fügen Sie den Paketnamen das Präfix com.google.cloud.spark.bigtable.repackaged hinzu. Verwenden Sie beispielsweise com.google.cloud.spark.bigtable.repackaged.com.google.cloud.bigtable.data.v2.BigtableDataClient anstelle von com.google.cloud.bigtable.data.v2.BigtableDataClient als Klassennamen.

Weitere Informationen zum Bigtable-Client für Java finden Sie im Hilfeartikel Bigtable-Client für Java.

Nächste Schritte