"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Spark BigQuery-Connector verwenden

Sie können den spark-bigquery-connector mit Managed Service for Apache Spark verwenden, um Daten aus BigQuery zu lesen und zu schreiben. In dieser Anleitung wird eine PySpark-Anwendung gezeigt, die die spark-bigquery-connector verwendet.

Connector-Version bestätigen

Unter Managed Service for Apache Spark-Laufzeitversionen finden Sie die BigQuery-Connector-Version, die in der Laufzeitversion Ihrer Batcharbeitslast oder interaktiven Sitzung installiert ist. Wenn der Connector nicht aufgeführt ist, lesen Sie den Abschnitt Connector für Anwendungen verfügbar machen.

Connector für Anwendungen verfügbar machen (falls erforderlich)

Der BigQuery-Connector ist in allen unterstützten Managed Service for Apache Spark-Laufzeitversionen installiert. Wenn Sie eine nicht unterstützte Laufzeitversion verwenden, in der der Connector (Spark runtime 1.0) nicht installiert wird, können Sie den Connector einer Anwendung auf eine der beiden folgenden Arten zur Verfügung stellen:

Verwenden Sie den Parameter jars, um auf eine Connector-JAR-Datei zu verweisen, wenn Sie eine Batcharbeitslast für Managed Service for Apache Spark senden oder eine interaktive Sitzung ausführen. Im folgenden Beispiel für einen Batch-Arbeitslast wird eine Connector-JAR-Datei angegeben. Eine Liste der verfügbaren Connector-JAR-Dateien finden Sie im GitHub-Repository GoogleCloudDataproc/spark-bigquery-connector.
- Beispiel für Google Cloud CLI:
```
gcloud dataproc batches submit pyspark \
    --region=REGION \
    --jars=spark-3.5-bigquery-version.jar \
    ... other args
```

Kosten berechnen

In dieser Anleitung werden kostenpflichtige Komponenten von Google Cloudverwendet, darunter:

Managed Service for Apache Spark
BigQuery
Cloud Storage

Der Preisrechner kann eine Kostenschätzung anhand Ihrer voraussichtlichen Nutzung generieren.

Neuen Cloud Platform-Nutzern steht gegebenenfalls eine kostenlose Testversion zur Verfügung.

Abrechnung konfigurieren

Standardmäßig wird das Projekt, das mit den Anmeldedaten oder dem Dienstkonto verbunden ist, für die API-Nutzung abgerechnet. Wenn Sie ein anderes Projekt in Rechnung stellen möchten, legen Sie das folgende Konfigurationsattribut fest: spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

Sie können diese Property auch einem Lese- oder Schreibvorgang hinzufügen, wie hier gezeigt: .option("parentProject", "<BILLED-GCP-PROJECT>").

PySpark-Batcharbeitslast für WordCount senden

In diesem Beispiel werden Daten aus BigQuery in einen Spark-DataFrame eingelesen und dann mit der Standard-Datenquellen-API einer Wortzählung unterzogen.

Der Connector schreibt die Wordcount-Ausgabe in BigQuery in der folgenden Reihenfolge von Vorgängen:

Puffert die Daten in temporären Dateien in Ihrem Cloud Storage-Bucket
Kopiert die Daten in einem Vorgang aus Ihrem Cloud Storage-Bucket in BigQuery
Löscht die temporären Dateien in Cloud Storage nach Abschluss des BigQuery-Ladevorgangs. Temporäre Dateien werden auch nach Beendigung der Spark-Anwendung gelöscht. Wenn das Löschen fehlschlägt, müssen Sie alle unerwünschten temporären Cloud Storage-Dateien löschen, die sich in der Regel in gs://BUCKET_NAME/.spark-bigquery-JOB_ID-UUID befinden.

Schritte zum Ausführen des Wordcount-Arbeitslast

Öffnen Sie ein lokales Terminal oder die Cloud Shell.
Erstellen Sie die wordcount_dataset mit dem bq-Befehlszeilentool in einem lokalen Terminal oder in Cloud Shell.
```
bq mk wordcount_dataset
```
Erstellen Sie einen Cloud Storage-Bucket mit der Google Cloud CLI.
```
gcloud storage buckets create gs://BUCKET_NAME
```
Ersetzen Sie BUCKET_NAME durch den Namen des Cloud Storage-Bucket, den Sie erstellt haben.

Erstellen Sie die Datei wordcount.py lokal in einem Texteditor, indem Sie den folgenden PySpark-Code kopieren.

#!/usr/bin/python
"""BigQuery I/O PySpark example."""
from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Cloud Storage bucket used by the connector for temporary BigQuery
# export data.
bucket = "BUCKET_NAME"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .load('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') \
  .save('wordcount_dataset.wordcount_output')

Senden Sie die PySpark-Batcharbeitslast:

gcloud dataproc batches submit pyspark wordcount.py \
    --region=REGION \
    --deps-bucket=BUCKET_NAME

Beispiel für Terminalausgabe:

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

Wenn Sie eine Vorschau der Ausgabetabelle in der Google Cloud Console aufrufen möchten, öffnen Sie die Seite BigQuery, wählen Sie die Tabelle wordcount_output aus und klicken Sie auf Vorschau.