pymongo MongoClient can not work in multiprocess?

Question

I am using pymongo 3.2, I want to use it in multiporcess:

client = MongoClient(JD_SEARCH_MONGO_URI, connect=False)
db = client.jd_search

with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    for jd in db['sample_data'].find():
        jdId = jd["jdId"]
        for cv in db["sample_data"].find():
            itemId = cv["itemId"]
            executor.submit(intersect_compute, jdId, itemId)
            #  print "done {} => {}".format(jdId, itemId)

but I get error:

UserWarning: MongoClient opened before fork. Create MongoClient with connect=False, or create client after forking. See PyMongo's documentation for details: http://api.mongodb.org/python/current/faq.html#using-pymongo-with-multiprocessing>

according to the documentation, I have set connect to False as you can see

Answer 1

You did exactly like in the documentation (URL in exception), but in the section Never do this.
p.s. I updated your code sample at the end of the comment.

Create connection to database inside each process:

# Each process creates its own instance of MongoClient.
def func():
    db = pymongo.MongoClient().mydb
    # Do something with db.

proc = multiprocessing.Process(target=func)
proc.start()

Never do this:

client = pymongo.MongoClient()

# Each child process attempts to copy a global MongoClient
# created in the parent process. Never do this.
def func():
  db = client.mydb
  # Do something with db.

proc = multiprocessing.Process(target=func)
proc.start()

What you need to change is to move database connection initialization to a fork of each process. As each of them will have own independent connection.

Your sample updated:

with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor:
    client = MongoClient(JD_SEARCH_MONGO_URI, connect=False)
    db = client.jd_search

    for jd in db['sample_data'].find():
        jdId = jd["jdId"]
        for cv in db["sample_data"].find():
            itemId = cv["itemId"]
            executor.submit(intersect_compute, jdId, itemId)
            #  print "done {} => {}".format(jdId, itemId)