PySpark MongoDB append all elements of an array from DataFrame

Question

I have a MongoDB collection that looks something like this:

{
    "_id" : { "customerName" : "Bob",  "customerPhone" : "123-456-7890"},
    "purchases": ["A", "B", "C", "D"]
}

Basically the _id is a pair of unique key about the customer and the purchases are the array of the items that the customer has bought.

I also have a PySpark DataFrame that I would like to push into this collection where it contains information that I would like to update this particular document.

df.write.format("com.mongodb.spark.sql.DefaultSource").mode("append") \
                .option("spark.mongodb.output.uri", "mongodb://localhost:27017/customer.purchases").save()

The problem is that if I am updating this document where I want to add new purchases for Bob it would only append the non-existing ones in purchases instead of appending all.

As a result what I ended up doing right now is that I would just have to call rdd.collect() to turn the whole thing into a list and not using the schema to convert it to DataFrame. Then insert everything one by one while checking if the key exists; which makes this part slow and requires a lot of memory when the query for RDD gets large.

For versions:

PySpark: 2.2 MongoDB: 3.0.15 Mongo Spark Connector: 2.2.1

Does anyone if there is anything that I can do to append all element in the array to MongoDB collection using dataframe? Also, if there is anything that I am missing or other things that I should do please let me know. Thanks!

user_2934579 · Answer

You need to change the data models or schema of your documents. The important part here being the _id key field. The field name _id is reserved for use as a primary key; its value must be unique in the collection, is immutable, and may be of any type other than an array.

In your case, the value of _id field is mutable, in fact this is what you're trying to update. As a suggestion, you may want to change this to:

{ "_id" : 
  "customerName" : "Bob",  
  "customerPhone" : "123-456-7890",
  "purchases": ["A", "B", "C", "D"]
}

You may utilise the default _id value of ObjectId as a unique identifier.

Once you have a unique identifier on _id field, let's talk about update operation. Since MongoDB Spark Connector v1.1+ (currently version 2.2), if a Dataframe contains an _id field during write, the data will be upsert-ed. Which means any existing documents with the same _id value will be updated and new documents without existing _id value in the collection will be inserted.

Bonus rounds:

You need to find a better schema as well for purchases field. Having an array length of undefined length may create issues in the future. i.e. Bob made 1000 purchases in a year.
Please update your MongoDB server version (version 3.0.x is from 2015), the current stable version is 3.4, with 3.6 being released in the next month.

PySpark MongoDB append all elements of an array from DataFrame

1 Answers