Updating 100s of Millions of Documents

Working with Data Drivers
,
4

Then you need to use the updateOne operation, also with an upsert within a BulkWrite operation.

Let me try to illustrate with an example in Python.

from datetime import datetime
from pprint import pprint

from faker import Faker
from pymongo import MongoClient, UpdateOne

fake = Faker()


def rand_host():
    return {
        "ip": fake.ipv4(),
        "ports": [rand_port(), rand_port()],
        "banners": [{"banner": "my first banner"}],
        "certificates": [{"cert": "cert 1.0"}],
        "first_scanned": datetime.now(),
        "last_scanned": datetime.now(),
        "headers": [{"header": "some header"}]
    }


def rand_port():
    return fake.pyint(min_value=1, max_value=65535)


if __name__ == '__main__':
    client = MongoClient()
    db = client.get_database('shodan')
    hosts = db.get_collection('hosts')

    # clean the db
    hosts.delete_many({})

    # init my hosts collection with one host
    hosts.create_index("ip", unique=True)
    hosts.insert_one(rand_host())

    print('Print the existing document in my collection:\n')
    init_doc = hosts.find_one()
    pprint(init_doc)
    init_ip = init_doc.get('ip')

    print('\nIP already known:', init_ip)

    print('\nNow I will try to insert 2 new hosts but the first one will have the same IP address than the one already in my collection.')

    hosts.bulk_write([
        # this first document will be updated. $setOnInsert won't do anything as it's not an insert.
        UpdateOne({'ip': init_ip},
                  {'$setOnInsert': {'first_scanned': datetime.now()},
                   '$addToSet': {'ports': {'$each': [rand_port(), rand_port()]},
                                 'banners': {'banner': 'my second banner'},
                                 "certificates": {"cert": "cert 2.0"},
                                 "headers": {"header": "some other header"}
                                 },
                   '$set': {'last_scanned': datetime.now()}
                   }, upsert=True),
        # this ip address doesn't exist in my collection so it's an insert.
        UpdateOne({'ip': fake.ipv4()},
                  {'$setOnInsert': {'first_scanned': datetime.now()},
                   '$addToSet': {'ports': {'$each': [rand_port(), rand_port()]},
                                 'banners': {'banner': 'my second banner'},
                                 "certificates": {"cert": "cert 2.0"},
                                 "headers": {"header": "some other header"}
                                 },
                   '$set': {'last_scanned': datetime.now()}
                   }, upsert=True)
    ], ordered=False)

    print('Final result in my hosts collection:\n')
    for doc in hosts.find():
        pprint(doc)
        print()

Console output:

Print the existing document in my collection:

{'_id': ObjectId('6006fec1f502dc4efd9da90c'),
 'banners': [{'banner': 'my first banner'}],
 'certificates': [{'cert': 'cert 1.0'}],
 'first_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 164000),
 'headers': [{'header': 'some header'}],
 'ip': '147.19.133.207',
 'last_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 164000),
 'ports': [29658, 6283]}

IP already known: 147.19.133.207

Now I will try to insert 2 new hosts but the first one will have the same IP address than the one already in my collection.
Final result in my hosts collection:

{'_id': ObjectId('6006fec1f502dc4efd9da90c'),
 'banners': [{'banner': 'my first banner'}, {'banner': 'my second banner'}],
 'certificates': [{'cert': 'cert 1.0'}, {'cert': 'cert 2.0'}],
 'first_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 164000),
 'headers': [{'header': 'some header'}, {'header': 'some other header'}],
 'ip': '147.19.133.207',
 'last_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 165000),
 'ports': [29658, 6283, 27037, 11895]}

{'_id': ObjectId('6006fec10ef49ae99654a648'),
 'banners': [{'banner': 'my second banner'}],
 'certificates': [{'cert': 'cert 2.0'}],
 'first_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 166000),
 'headers': [{'header': 'some other header'}],
 'ip': '79.222.50.147',
 'last_scanned': datetime.datetime(2021, 1, 19, 16, 46, 9, 166000),
 'ports': [47533, 38525]}

As you can see in my 2 final documents, the one which already existed in my collection has been updated correctly with the new ports, certificates and headers. The first_scanned field is untouched as it’s not an insert operation (thanks to the $setOnInsert).

The second one in brand new as it’s IP address was never seen before.

If you don’t want to update the document but rather completely update it, you could use replaceOne maybe? Depends what you want to do exactly.
You could replace some values also with $set instead of using the $addToSet or $push that work with arrays and will just append the new values in the array.

I hope it helps and this is what you needed :slight_smile:.

Cheers,
Maxime.

PS EDIT: I added ordered=False on the Bulk Op as it’s more efficient in your scenario and you won’t stop on the first error you might have.

4 Likes