<jupyter_start><jupyter_text>Upsert DataFrame to MongoDB>MongoDB is a rich document-oriented NoSQL database. I spent some time diving into it. In this post I will share how to 'upsert' the Pandas DataFrame against the MongoDB- toc: true- comments: true- branch: master- badges: true- categories: [NoSQL, Python]- image: images/upsert.png  RequirementLet's say we have a DataFrame. We want to 'upsert'(Insert/update) the MongoDB collection with rows in DataFrame. More specifically, for the rows that don't exist we do the creation, otherwise the update should be performed.<jupyter_code>#hide
from pymongo import MongoClient, UpdateOne, collection
from pandas import DataFrame
from numpy import where
from datetime import datetime
    
CONN_STR = 'mongodb+srv://fin-market.x5tm4.mongodb.net/myFirstDatabase?authSource=%24external&authMechanism=MONGODB-X509&tls=true&tlsCertificateKeyFile=../../../drivers/X509-cert-fin-market.pem'
#collapse_output
def show_collection(collection):
    display(DataFrame([row for row in collection.find()]))<jupyter_output><empty_output><jupyter_text>Setup Cloud MongoDB for FreeSetup a free cloud based MongoDB database [here](https://www.mongodb.com/cloud/atlas/lp/try2?utm_source=google&utm_campaign=gs_apac_hong_kong_search_core_brand_atlas_desktop&utm_term=mongodb%20atlas&utm_medium=cpc_paid_search&utm_ad=e&utm_ad_campaign_id=12212624344&adgroup=115749709143&gclid=Cj0KCQiAqbyNBhC2ARIsALDwAsApK8irJVZnesxewSitv8kTagWactqhvZPG4gpz5CKx0JV3Fh1pEIsaAttLEALw_wcB).Connect to the database and create the 'test' database and 'employee' collection.<jupyter_code>DB_NAME = 'test'
COLLECTION_NAME = 'employee'
client = MongoClient(CONN_STR)
client.drop_database(DB_NAME) # clear the database
db = client[DB_NAME] # switch database
collection = db[COLLECTION_NAME] # get the collection
#collapse_output
now = datetime.now()
emplyee = [['user1',25,'male', now],['user2',55,'male', now],['user3',43,'male', now]]
df_emplyee = DataFrame(emplyee, columns=['name','age','sex', 'lastModifiedAt'])
df_emplyee = df_emplyee[['name','age','sex']]<jupyter_output><empty_output><jupyter_text>Prepare a test DataFrame df_emplyee that contains three columns:- name- age- sex<jupyter_code>df_emplyee<jupyter_output><empty_output><jupyter_text>Once the records are inserted to the collection, we get the \_id column for every row. ['\_id' column is a potentially a good shard key used by MongoDB cluster. ](https://www.mongodb.com/blog/post/on-selecting-a-shard-key-for-mongodb)<jupyter_code>collection.insert_many(df_emplyee.to_dict("records"))
show_collection(collection)<jupyter_output><empty_output><jupyter_text>UpdateLet's do the update. Change the age of user1 to 32 with [update_one](https://pymongo.readthedocs.io/en/stable/api/pymongo/collection.html) method.- filter parameter get the rows that need update.- [\$set](https://docs.mongodb.com/manual/reference/operator/update/set/) flag is to update the columns specified.- [\$currentDate](https://docs.mongodb.com/manual/reference/operator/update/currentDate/) here is to update lastModifiedAt column with the current time.<jupyter_code>myquery = { "name": "user1" }
newvalues = { "$set": { "age": "32" }, "$currentDate": {"lastModifiedAt": { "$type": "date" }} }
res = collection.update_one(filter=myquery, update=newvalues)
show_collection(collection)<jupyter_output><empty_output><jupyter_text>UpsertAs the name suggests 'upsert' means update/insert records based on the specified filters. For intansce, in the following first example, 'user1' exists in the collection. The 'upsert' performs the update. In the second example, user4 doesn't exist and it is inserted to the collection.<jupyter_code>collection.update_one({"name":"user1"}, 
                      {"$set":{"age":32},
                              "$currentDate": 
                              {"lastModifiedAt": { "$type": "date" }}
                      }, 
                      upsert=True)
show_collection(collection)
collection.update_one({"name":"user4"}, 
                      {"$set":{"age":32}, 
                              "$setOnInsert":{"sex":"female"},
                              "$currentDate":{"lastModifiedAt": { "$type": "date" }}
                      }, 
                      upsert=True)
show_collection(collection)<jupyter_output><empty_output><jupyter_text>Bulk updatesObviously it's inefficient if the updates/inserts are performed one by one. To improve this, we firstly collect all the 'UpdateOne' operations and perform the updates with ['bulk_write'](https://docs.mongodb.com/manual/reference/method/db.collection.bulkWrite/) like the following.<jupyter_code>df_emplyee = df_emplyee.append({'name':'user5','age': 65, 'sex':'male'},ignore_index=True)


updates = []
df_emplyee.apply(
    lambda row: updates.append(
        UpdateOne(
            {"name": row.get("name")}, 
            {"$set": row.to_dict(), 
                     "$currentDate":{"lastModifiedAt": { "$type": "date" }}
            }, 
            upsert=True
        )),
    axis=1)
collection.bulk_write(updates)
show_collection(collection)<jupyter_output><empty_output><jupyter_text>DataFrame UpsertBack to our initial requirement, we need to perform the 'Upsert' on individual DataFrame row. I created a function df_upsert for this.<jupyter_code>def df_upsert(df:DataFrame, collection, keys:[]):
    def row_query(row, keys ):
        res = {}
        for key in keys:
            res[key] = row.get(key)
        return res
    updates = []
    df_emplyee.apply(
        lambda row: updates.append(
            UpdateOne(
                row_query(row, keys), 
                {'$set': row.to_dict(),
                    "$currentDate":{"lastModifiedAt": { "$type": "date" }}
                }, 
                upsert=True)), 
            axis=1
    )
    collection.bulk_write(updates)<jupyter_output><empty_output><jupyter_text>Let's do some testing. Some updates are applied to the df_emplyee. Here are the difference between the df_emplyee and the MongoDB collection. As we see, user6 is added and user1's age is changed to 20 from 25. Please note user4 doesn't exist in the df_emplyee, it will not be touched in the collection.<jupyter_code>df_emplyee.loc[0,'age'] = 20
df_emplyee = df_emplyee.append({'name':'user6','age': 37, 'sex':'female'},ignore_index=True)

show_collection(collection)
display(df_emplyee)<jupyter_output><empty_output><jupyter_text>Once we perform the 'df_upsert', we get the expected results.<jupyter_code>df_upsert(df_emplyee, collection, ['name'])
show_collection(collection)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12122132766216011, 'fraction_numerical': 0.0130639526051952, 'mean_word_length': 4.331174089068826, 'pattern_counts': {'":': 28, '<': 41, '<?xml version=': 0, '>': 42, 'https://': 6, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '469200', 'license': 'Apache-2.0', 'n_tokens_mistral': 2165, 'n_tokens_neox': 2037, 'n_words': 489, 'path': '_notebooks/2021-12-09-mogodb_01.ipynb', 'repo_name': 'stevenchen521/blogging', 'size': '6583'}