"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 utilise la technologie IA pour traduire le contenu dans votre langue préférée. Les traductions générées par IA peuvent contenir des erreurs.

Utiliser le connecteur Spark BigQuery

Vous pouvez utiliser le connecteur spark-bigquery avec Managed Service pour Apache Spark pour lire et écrire des données depuis et vers BigQuery. Ce tutoriel présente une application PySpark qui utilise spark-bigquery-connector.

Confirmer la version du connecteur

Consultez les versions d'exécution de Managed Service pour Apache Spark pour déterminer la version du connecteur BigQuery installée dans la version d'exécution de votre charge de travail par lot ou de votre session interactive. Si le connecteur n'est pas listé, consultez Rendre le connecteur disponible pour les applications.

Rendre le connecteur disponible pour les applications (si nécessaire)

Le connecteur BigQuery est installé dans toutes les versions d'exécution Managed Service pour Apache Spark compatibles. Si vous utilisez une version d'exécution non compatible qui n'installe pas le connecteur (Spark runtime 1.0), vous pouvez rendre le connecteur disponible pour une application de deux manières différentes :

Utilisez le paramètre jars pour pointer vers un fichier jar de connecteur lorsque vous envoyez une charge de travail par lot Managed Service pour Apache Spark ou exécutez une session interactive. L'exemple de charge de travail par lot suivant spécifie un fichier jar de connecteur (pour obtenir la liste des fichiers jar de connecteur disponibles, consultez le dépôt GoogleCloudDataproc/spark-bigquery-connector sur GitHub).
- Exemple de Google Cloud CLI :
```
gcloud dataproc batches submit pyspark \
    --region=REGION \
    --jars=spark-3.5-bigquery-version.jar \
    ... other args
```

Calculer les coûts

Ce tutoriel fait appel à des composants payants de Google Cloud, y compris :

Managed Service pour Apache Spark
BigQuery
Cloud Storage

Utilisez le Simulateur de coût pour générer une estimation des coûts en fonction de votre utilisation prévue.

Les nouveaux utilisateurs de Cloud Platform peuvent bénéficier d'un essai sans frais.

Configurer la facturation

Par défaut, le projet qui est facturé pour l'utilisation de l'API est le projet associé aux identifiants ou au compte de service. Pour facturer un autre projet, définissez la propriété de configuration suivante : spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

Vous pouvez également ajouter cette propriété à une opération de lecture ou d'écriture, comme suit :.option("parentProject", "<BILLED-GCP-PROJECT>").

Envoyer une charge de travail par lot de décompte de mots PySpark

Cet exemple lit les données de BigQuery dans un DataFrame Spark pour effectuer un décompte de mots à l'aide de l'API de source de données standard.

Le connecteur écrit le résultat du décompte de mots dans BigQuery dans la séquence d'opérations suivante :

Met en mémoire tampon les données dans des fichiers temporaires de votre bucket Cloud Storage
Copie les données en une seule opération depuis votre bucket Cloud Storage vers BigQuery
Supprime les fichiers temporaires dans Cloud Storage une fois l'opération de chargement BigQuery terminée (les fichiers temporaires sont également supprimés une fois l'application Spark terminée). Si la suppression échoue, vous devrez supprimer tous les fichiers Cloud Storage temporaires indésirables, qui sont généralement placés dans gs://BUCKET_NAME/.spark-bigquery-JOB_ID-UUID.

Procédure pour exécuter la charge de travail WordCount

Ouvrez un terminal local ou Cloud Shell.
Créez le wordcount_dataset avec l'outil de ligne de commande bq dans un terminal local ou dans Cloud Shell.
```
bq mk wordcount_dataset
```
Créez un bucket Cloud Storage avec la Google Cloud CLI.
```
gcloud storage buckets create gs://BUCKET_NAME
```
Remplacez BUCKET_NAME par le nom du bucket Cloud Storage que vous avez créé.

Créez le fichier wordcount.py en local dans un éditeur de texte en copiant le code PySpark suivant.

#!/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')

Envoyez la charge de travail par lot PySpark :

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

Exemple de résultat du terminal :

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

Pour prévisualiser la table de sortie dans la console Google Cloud , ouvrez la page BigQuery, sélectionnez la table wordcount_output, puis cliquez sur Aperçu.