E-MapReduce:Read from and write to MongoDB

Step 1: Obtain MongoDB connector JARs and upload to OSS

1. Based on your Spark and MongoDB versions, download the required dependencies from Maven. For more information, see the official MongoDB Spark Connector documentation. This example uses MongoDB 5.0.1, which requires the following JAR packages:

   • 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. Upload the downloaded JAR packages to Alibaba Cloud Object Storage Service (OSS). For more information, see simple upload.

Step 2: Create a network connection

EMR Serverless Spark requires a network connection to your MongoDB database. For more information about network connections, see Connect EMR Serverless Spark to other VPCs.

Step 3: Read from a MongoDB collection

1. Create a notebook session. For more information, see Manage notebook sessions.

   When you create the session, select a supported Engine Version and the Normal Network Connection from Step 2. Then, in the Spark Configuration section, add the following parameters to load the MongoDB Spark Connector.

   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

   Replace the placeholder values to match your environment.

   Parameter	Description	Example
   spark.mongodb.write.connection.uri	The connection URI for reading from and writing to MongoDB. <mongodb_host>: The IP address or hostname of your MongoDB service. 27017: The default port for MongoDB.	mongodb://192.168.x.x:27017
   spark.mongodb.read.connection.uri
   spark.emr.serverless.user.defined.jars	A comma-separated list of external dependencies for Spark.	A JAR package uploaded to OSS. Example: oss://<yourBucketname>/spark/mongodb/mongo-spark-connector_2.12-10.4.1.jar.

2. On the Development page, create a Python > notebook task, and then select the notebook session that you created.

   For more information, see Manage notebook sessions.

3. Copy the following code into a notebook cell, modify the parameters, and then click Run.

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

   The following table describes the parameters.

   Parameter	Description
   <yourDatabase>	The name of your MongoDB database. Example: mongo_table.
   <yourCollection>	The name of your MongoDB collection. Example: MongoCollection.

   If the data is returned successfully, the configuration is correct. The expected output is shown below:

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

Step 4: Write to a MongoDB collection

Copy the following code into a new cell in your notebook, modify the parameters, and then click 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()
              

The following output from df.show() displays the data before it is written to MongoDB. A successful execution confirms that the configuration is correct and that the data has been appended to your collection.

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