E-MapReduce:Lire et écrire dans MongoDB

Étape 1 : Obtenir les fichiers JAR du connecteur MongoDB et les télécharger vers OSS

1. Selon vos versions de Spark et de MongoDB, téléchargez les dépendances requises depuis Maven. Pour plus d'informations, consultez la documentation officielle du connecteur Spark pour MongoDB. Cet exemple utilise MongoDB 5.0.1, qui nécessite les packages JAR suivants :

   • mongo-spark-connector_2.12-10.4.1.jar

   • mongodb-driver-core-5.0.1.jar

   • mongodb-driver-sync-5.0.1.jar

   • bson-5.0.1.jar

2. Téléversez les packages JAR téléchargés vers Alibaba Cloud Object Storage Service (OSS). Pour plus d'informations, consultez la section téléchargement simple.

Étape 2 : Créer une connexion réseau

EMR Serverless Spark nécessite une connexion réseau à votre base de données MongoDB. Pour plus d'informations, consultez la section Connecter EMR Serverless Spark à d'autres VPC.

Étape 3 : Lire une collection MongoDB****

1. Créez une session de notebook. Pour plus d'informations, consultez la section Gérer les sessions de notebook.

   Lors de la création de la session, sélectionnez une Engine Version prise en charge ainsi que la Normal Network Connection configurée à l'étape 2. Ensuite, dans la section Spark Configuration, ajoutez les paramètres suivants pour charger le connecteur Spark pour MongoDB.

   spark.mongodb.write.connection.uri                mongodb://<mongodb_host>:27017
   spark.mongodb.read.connection.uri                 mongodb://<mongodb_host>:27017
   spark.emr.serverless.user.defined.jars            oss://<bucketname>/path/to/mongo-spark-connector_2.12-10.4.1.jar,oss://<bucketname>/path/to/mongodb-driver-core-5.0.1.jar,oss://<bucketname>/path/to/mongodb-driver-sync-5.0.1.jar,oss://<bucketname>/path/to/bson-5.0.1.jar

   Remplacez les valeurs d'espace réservé par celles correspondant à votre environnement.

   Paramètre	Description	Exemple
   spark.mongodb.write.connection.uri	URI de connexion pour la lecture et l'écriture dans MongoDB. <mongodb_host> : adresse IP ou nom d'hôte de votre service MongoDB. 27017 : port par défaut de MongoDB.	mongodb://192.168.x.x:27017
   spark.mongodb.read.connection.uri
   spark.emr.serverless.user.defined.jars	Liste séparée par des virgules des dépendances externes pour Spark.	Package JAR téléversé vers OSS. Exemple : oss://<yourBucketname>/spark/mongodb/mongo-spark-connector_2.12-10.4.1.jar.

2. Sur la page Development, créez une tâche Python > notebook, puis sélectionnez la session de notebook que vous avez créée.

   Pour plus d'informations, consultez la section Gérer les sessions de notebook.

3. Copiez le code suivant dans une cellule de notebook, modifiez les paramètres, puis cliquez sur Run.

   df = spark.read \
       .format("mongodb") \
       .option("database", "<yourDatabase>") \
       .option("collection", "<yourCollection>") \
       .load()
   df.printSchema()
   df.show()
   

   Le tableau suivant décrit les paramètres :

   Paramètre	Description
   <yourDatabase>	Nom de votre base de données MongoDB. Exemple : mongo_table.
   <yourCollection>	Nom de votre collection MongoDB. Exemple : MongoCollection.

   Une requête réussie confirme que la configuration est correcte. Sortie attendue :

   root
    |-- _id: string (nullable = true)
    |-- age: integer (nullable = true)
    |-- city: string (nullable = true)
    |-- name: string (nullable = true)
   +--------------------+---+-----------+-------+
   |                 _id|age|       city|   name|
   +--------------------+---+-----------+-------+
   |67e65cb60ccf7f94a...| 30|Los Angeles|    Bob|
   |67e65cb60ccf7f94a...| 35|    Chicago|Charlie|
   +--------------------+---+-----------+-------+

Étape 4 : Écrire dans une collection MongoDB****

Copiez le code suivant dans une nouvelle cellule de votre notebook, modifiez les paramètres, puis cliquez sur Run.

from pyspark.sql import Row
data = [
    Row(name="Sam", age=25, city="New York"),
    Row(name="Charlie", age=35, city="Chicago")
]
df = spark.createDataFrame(data)
df.show()
df.write \
    .format("mongodb") \
    .option("database", "<yourDatabase>") \
    .option("collection", "<yourCollection>") \
    .mode("append") \
    .save()
              

La sortie de df.show() affiche les données avant leur écriture dans MongoDB. Une exécution réussie confirme que la configuration est correcte et que les données ont été ajoutées à votre collection.

+-------+---+--------+
|   name|age|    city|
+-------+---+--------+
|    Sam| 25|New York|
|Charlie| 35| Chicago|
+-------+---+--------+