Bigtable Spark-Connector verwenden
Mit dem Bigtable Spark-Connector können Sie Daten in Bigtable lesen und schreiben. Sie können Daten mit Spark SQL und DataFrames aus Ihrer Spark-Anwendung lesen. Die folgenden Bigtable-Vorgänge werden mit dem Bigtable Spark-Connector unterstützt:
- Daten schreiben
- Daten lesen
- Neue Tabelle erstellen
In diesem Dokument erfahren Sie, wie Sie eine Spark SQL-DataFrames-Tabelle konvertieren. in eine Bigtable-Tabelle hochladen und dann eine JAR-Datei kompilieren und erstellen. um einen Spark-Job zu senden.
Supportstatus für Spark und Scala
Der Bigtable Spark-Connector unterstützt nur die Version Scala 2.12 und die folgenden Spark-Versionen:
Der Bigtable Spark-Connector unterstützt die folgenden Dataproc-Versionen:
- 1.5 Image-Version-Cluster
- 2.0-Image-Version-Cluster
- 2.1 Cluster mit Image-Version
- 2.2 image-version cluster
- Dataproc Serverless-Laufzeitversion 1.0
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
Um die Berechtigungen zu erhalten, die Sie zur Verwendung des Bigtable Spark-Connectors benötigen, bitten Sie Ihren Administrator, Ihnen folgende IAM-Rollen für Ihr Projekt:
-
Bigtable-Administrator (
roles/bigtable.admin
)(optional): können Sie Daten lesen oder schreiben und eine neue Tabelle erstellen. -
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 for Spark im Vergleich zu Dataproc in der Compute Engine.
Connector-JAR-Datei 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,2.12
als einzige unterstützte Version undCONNECTOR_VERSION
durch die Connector-Version, die Sie verwenden möchten.Verwenden Sie für Dataproc-Cluster oder serverlose Option die neueste JAR-Datei als Artefakt, das in Ihre 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 den 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 schreibgeschützte Jobs können Sie das serverlose Computing Data Boost (Vorabversion) verwenden, das können Sie eine Beeinträchtigung der Anwendungsbereitstellungscluster vermeiden. Dein Spark Anwendung muss Version 1.1.0 oder höher des Spark-Connectors verwenden, um die Anwendung nutzen zu können. Daten-Boost.
Wenn Sie Data Boost verwenden möchten, müssen Sie ein Data Boost-Anwendungsprofil erstellen und dann
Geben Sie die Anwendungsprofil-ID für spark.bigtable.app_profile.id
Spark an
Option beim Hinzufügen der Bigtable-
in Ihre Spark-Anwendung ein. Wenn Sie bereits ein App-Profil für Ihre Spark-Lesejobs erstellt haben und es weiter verwenden möchten, ohne den Anwendungscode zu ändern, können Sie das App-Profil in ein Data Boost-App-Profil konvertieren. Weitere Informationen finden Sie unter App konvertieren
Profil.
Weitere Informationen finden Sie in der Übersicht über Bigtable Data Boost.
Für Jobs mit Lese- und Schreibvorgängen können Sie den Cluster Ihrer Instanz verwenden. Knoten für die Berechnung. Dazu geben Sie in Ihrer Anfrage ein Standard-Anwendungsprofil an.
App-Profil für die Verwendung auswählen 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 hierzu finden Sie unter App-Profile 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 vor dem Schreiben in Bigtable eine neue Tabelle zu erstellen. |
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 true fest, um das einfache Filtern von Zeilenschlüsseln auf der Serverseite zuzulassen. Das Filtern nach zusammengesetzten Zeilenschlüsseln wird clientseitig 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 das gesamte Zeitlimit für einen Leseversuch 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 das Zeitlimit für einen Änderungsversuch von Zeilen fest, der einer DataFrame-Partition im Bigtable-Client für Java entspricht. |
spark.bigtable.batch.mutate.size |
Nein | 100 |
Legen Sie als Wert 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 String-JSON-Format macht das Datenformat mit Bigtable kompatibel. Sie können das JSON-Format mit der Option .option("catalog", catalog_json_string)
in Ihren 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
und 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 | location | |||||||
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"}
}
}
"""
Im JSON-Format werden folgende Schlüssel und Werte verwendet:
Katalogschlüssel | Katalogwert | JSON-Format |
---|---|---|
table | Name der Bigtable-Tabelle. | "table":{"name":"t1"} Falls die Tabelle nicht vorhanden ist, verwenden Sie .option("spark.bigtable.create.new.table", "true") , um sie zu erstellen. |
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 . Zusammengesetzte Schlü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 in die entsprechende Bigtable-Spaltenfamilie ("cf" ) und den 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.
Darüber hinaus können Sie Avro zum Serialisieren komplexer
z. B. ArrayType
. Weitere Informationen finden Sie unter Komplexe Daten serialisieren
Typen mit Apache Avro
In Bigtable schreiben
Verwenden Sie die Funktion .write()
und die unterstützten Optionen, um Ihre Daten in Bigtable zu schreiben.
Java
Der folgende Code aus dem GitHub-Repository verwendet Java und Maven, um in Bigtable zu schreiben.
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
Der folgende Code aus dem GitHub-Repository verwendet Python, um 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
Prüfen Sie mit der Funktion .read()
, 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, Dataproc Serverless oder einer lokalen Spark-Instanz ausgeführt wird. Sie können die JAR-Datei lokal kompilieren und dann zum Senden eines Jobs verwenden. 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:
Kompilieren Sie die JAR-Datei.
Maven
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>
Fügen Sie der Datei
pom.xml
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>
Führen Sie den Befehl
mvn clean install
aus, um eine JAR-Datei zu generieren.
sbt
Fügen Sie der Datei
build.sbt
die Abhängigkeitspark-bigtable
hinzu:libraryDependencies += "com.google.cloud.spark.bigtable" % "spark-bigtable_SCALA_VERSION" % "0.1.0{""}}"
Fügen Sie Ihrer
project/plugins.sbt
- oderproject/assembly.sbt
-Datei dassbt-assembly
-Plug-in hinzu, um eine Uber-JAR-Datei zu erstellen.addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")
Führen Sie den Befehl
sbt clean assembly
aus, um die JAR-Datei zu generieren.
Gradle
Fügen Sie der Datei
build.gradle
die Abhängigkeitspark-bigtable
hinzu.dependencies { implementation group: 'com.google.cloud.bigtable', name: 'spark-bigtable_SCALA_VERSION', version: '0.1.0' }
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' }
Weitere Informationen zur Konfiguration und zur 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 zum Hochladen der Abhängigkeiten von Spark-Arbeitslasten.
- 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 Dateispark-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-Einrichtung 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.
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
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 mit einer Dataproc Serverless-Instanz Daten 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
Lokales Spark
Verwenden Sie die heruntergeladene JAR-Datei und erstellen Sie einen Spark-Job, der Daten mit einer lokalen Spark-Instanz in Bigtable liest und schreibt. Sie können den Spark-Job auch über den Bigtable-Emulator einreichen.
Bigtable-Emulator verwenden
Wenn Sie sich für den Bigtable-Emulator entscheiden, gehen Sie so vor:
Verwenden Sie den folgenden Befehl, um den Emulator zu starten:
gcloud beta emulators bigtable start
Standardmäßig wählt der Emulator
localhost:8086
aus.Legen Sie die Umgebungsvariable
BIGTABLE_EMULATOR_HOST
fest:export BIGTABLE_EMULATOR_HOST=localhost:8086
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 überprü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, der nur bestimmte Zeilen liest. Einfache Filter wie ==
, <=
und startsWith
auf die Zeilenschlüsselspalte werden serverseitig angewendet, um einen Scan der gesamten Tabelle zu vermeiden. Filter für zusammengesetzte Zeilenschlüssel oder komplexe Filter wie der LIKE
-Filter für die Zeilenschlüsselspalte werden clientseitig angewendet.
Wenn Sie große Tabellen lesen, empfehlen wir die Verwendung einfacher Zeilenschlüsselfilter, um einen Scan der gesamten Tabelle zu vermeiden. Die folgende Beispielanweisung zeigt, wie mit einem einfachen Filter gelesen wird. Achten Sie darauf, dass Sie im Spark-Filter den Namen der DataFrame-Spalte verwenden, die in den Zeilenschlüssel konvertiert wird:
dataframe.filter("id == 'some_id'").show()
Wenn Sie einen Filter anwenden, verwenden Sie den Namen der DataFrame-Spalte anstelle des Bigtable-Tabellenspaltennamens.
Komplexe Datentypen mit Apache Avro serialisieren
Der Bigtable Spark-Connector unterstützt die Verwendung von Apache Avro, um komplexe Spark SQL-Typen wie ArrayType
, MapType
oder StructType
zu serialisieren. 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 beim Lesen oder Schreiben in Bigtable .option("avroSchema", avroSchemaString)
verwenden, um das zu dieser Spalte gehörende Avro-Schema im Stringformat anzugeben. 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 standardmäßig innerhalb des Connectors aktiviert. Weitere Informationen zum Aufrufen und Interpretieren 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 unter Bigtable-Client für Java.
Nächste Schritte
- Informationen zum Optimieren Ihres Spark-Jobs in Dataproc
- Verwenden Sie Klassen aus dem Bigtable-Client für Java mit dem Bigtable Spark-Connector.