call custom python function on every document in a collection Mongo DB

Question

I want to call a custom python function on some existing attribute of every document in the entire collection and store the result as a new key-value pair in that (same) document. May I know if there's any way to do that (since each call is independent of others) ?

I noticed cursor.forEach but can't it be done just using python efficiently ?

A simple example would be to split the string in text and store the no. of words as a new attribute.

def split_count(text):
    # some complex preprocessing...

    return len(text.split())

# Need something like this...
db.collection.update_many({}, {'$set': {"split": split_count('$text') }}, upsert=True)

But it seems like setting a new attribute in a document based on the value of another attribute in the same document is not possible this way yet. This post is old but the issues seem to be still open.

can you post some example code of what you're wanting to do? — Steve Rossiter, Jun 13 '16 at 10:51
@SteveRossiter uhh.. yes & no. It solves the current task (splitting the text) but I also want to call _any_ python function which makes me difficult to stick only to _JS_ approach. Also doing it _JS_ messes up some of my attribute values. Please see the answer below. :) — Kevad, Jun 24 '16 at 08:41

Kevad · Accepted Answer · 2016-06-24T08:44:25.880

I found a way to call any custom python function on a collection using parallel_scan in PyMongo.

def process_text(cursor):
    for row in cursor.batch_size(200):
        # Any complex preprocessing here...
        split_text = row['text'].split()

        db.collection.update_one({'_id': row['_id']}, 
                                 {'$set': {'split_text': split_text, 
                                           'num_words': len(split_text) }},
                                 upsert=True)


def preprocess(num_threads=4):

    # Get up to max 'num_threads' cursors.
    cursors = db.collection.parallel_scan(num_threads)
    threads = [threading.Thread(target=process_text, args=(cursor,)) for cursor in cursors]

    for thread in threads:
        thread.start()

    for thread in threads:
        thread.join()

This is not really faster than cursor.forEach (but not that slow either), but it helps me execute any arbitrarily complex python code and save the results from within Python itself.

Also if I have an array of ints in one of the attributes, doing cursor.forEach converts them to floats which I don't want. So I preferred this way.

But I would be glad to know if there're any better ways than this :)

Since Python doesn't support true parallelism, it can only be as fast as `cursor.forEach` and should be slower in the usual case. — Niklas R, Jun 26 '17 at 11:18

score 0 · Answer 2 · edited May 23 '17 at 11:58

It is quite unlikely that it will ever be efficient to do this kind of thing in python. This is because the document would have to make a round trip and go through the python function on the client machine.

In your example code, you are passing the result of a function to a mongodb update query, which won't work. You can't run any python code inside mongodb queries on the db server.

As the answer to you linked question suggests, this type of action has to be performed in the mongo shell. e.g:

db.collection.find().snapshot().forEach(
    function (elem) {
        splitLength = elem.text.split(" ").length
        db.collection.update(

            {
                _id: elem._id
            },
            {
                $set: {
                    split: splitLength 
                }
            }
        );
    }
);

call custom python function on every document in a collection Mongo DB

2 Answers2