title: Parallelizing loading data from MongoDB into python
body: <p>All documents in my collection in MongoDB have the same fields. My goal is to load them into Python into <code>pandas.DataFrame</code> or <code>dask.DataFrame</code>.</p>
  
  <p>I'd like to speedup the loading procedure by parallelizing it. My plan is to spawn several processes or threads. Each process would load a chunk of a collection, then these chunks would be merged together.</p>
  
  <p>How do I do it correctly with MongoDB?</p>
  
  <p>I have tried similar approach with PostgreSQL. My initial idea was to use <code>SKIP</code> and <code>LIMIT</code> in SQL queries. It has failed, since each cursor, opened for each particular query, started reading data table from the beginning and just skipped specified amount of rows. So I had to create additional column, containing record numbers, and specify ranges of these numbers in queries.</p>
  
  <p>On the contrary, MongoDB assigns unique ObjectID to each document. However, I've found that it is impossible to subtract one ObjectID from another, they can be only compared with ordering operations: less, greater and equal.</p>
  
  <p>Also, <code>pymongo</code> returns the cursor object, that supports indexing operation and has some methods, seeming useful for my task, like <code>count</code>, <code>limit</code>.</p>
  
  <p>MongoDB connector for Spark accomplishes this task somehow. Unfortunately, I'm not familiar with Scala, therefore, it's hard for me to find out how they do it.</p>
  
  <p>So, what is the correct way for parallel loading data from Mongo into python?</p>
  
  <p>up to now, I've come to the following solution:</p>
  
  <pre><code>import pandas as pd
  import dask.dataframe as dd
  from dask.delayed import delayed
  
  # import other modules.
  
  collection = get_mongo_collection()
  cursor = collection.find({ })
  
  def process_document(in_doc):
      out_doc = # process doc keys and values
      return pd.DataFrame(out_doc)
  
  df = dd.from_delayed( (delayed(process_document)(d) for d in cursor) )
  </code></pre>
  
  <p>However, it looks like <code>dask.dataframe.from_delayed</code> internally creates a list from passed generator, effectively loading all collection in a single thread.</p>
  
  <p><em>Update</em>. I've found <a href="https://docs.mongodb.com/manual/reference/method/cursor.skip/" rel="noreferrer">in docs</a>, that <code>skip</code> method of <code>pymongo.Cursor</code> starts from beginning of a collection too, as PostgreSQL. The same page suggests using pagination logic in the application. Solutions, that I've found so far, use sorted <code>_id</code> for this. However, they also store last seen <code>_id</code>, that implies that they also work in a single thread.</p>
  
  <p><em>Update2</em>. I've found the code of the partitioner in the official MongoDb Spark connector: <a href="https://github.com/mongodb/mongo-spark/blob/7c76ed1821f70ef2259f8822d812b9c53b6f2b98/src/main/scala/com/mongodb/spark/rdd/partitioner/MongoPaginationPartitioner.scala#L32" rel="noreferrer">https://github.com/mongodb/mongo-spark/blob/7c76ed1821f70ef2259f8822d812b9c53b6f2b98/src/main/scala/com/mongodb/spark/rdd/partitioner/MongoPaginationPartitioner.scala#L32</a></p>
  
  <p>Looks like, initially this partitioner reads the key field from all documents in the collection and calculates ranges of values.</p>
  
  <p><em>Update3</em>: My incomplete solution. </p>
  
  <p>Doesn't work, gets the exception from pymongo, because dask seems to incorrectly treat the <code>Collection</code> object:</p>
  
  <pre><code>/home/user/.conda/envs/MBA/lib/python2.7/site-packages/dask/delayed.pyc in &lt;genexpr&gt;(***failed resolving arguments***)
       81         return expr, {}
       82     if isinstance(expr, (Iterator, list, tuple, set)):
  ---&gt; 83         args, dasks = unzip((to_task_dask(e) for e in expr), 2)
       84         args = list(args)
       85         dsk = sharedict.merge(*dasks)
  
  /home/user/.conda/envs/MBA/lib/python2.7/site-packages/pymongo/collection.pyc in __next__(self)
     2342 
     2343     def __next__(self):
  -&gt; 2344         raise TypeError("'Collection' object is not iterable")
     2345 
     2346     next = __next__
  
  TypeError: 'Collection' object is not iterable
  </code></pre>
  
  <p>What raises the exception:</p>
  
  <pre><code>def process_document(in_doc, other_arg):
      # custom processing of incoming records
      return out_doc
  
  def compute_id_ranges(collection, query, partition_size=50):
      cur = collection.find(query, {'_id': 1}).sort('_id', pymongo.ASCENDING)
      id_ranges = [cur[0]['_id']]
      count = 1
      for r in cur: 
          count += 1
          if count &gt; partition_size:
              id_ranges.append(r['_id'])
              count = 0
      id_ranges.append(r['_id'])
      return zip(id_ranges[:len(id_ranges)-1], id_ranges[1: ])    
  
  
  def load_chunk(id_pair, collection, query={}, projection=None):
      q = query
      q.update( {"_id": {"$gte": id_pair[0], "$lt": id_pair[1]}} )
      cur = collection.find(q, projection)
  
      return pd.DataFrame([process_document(d, other_arg) for d in cur])
  
  
  def parallel_load(*args, **kwargs):
      collection = kwargs['collection']
      query = kwargs.get('query', {})
      projection = kwargs.get('projection', None)
  
      id_ranges = compute_id_ranges(collection, query)
  
      dfs = [ delayed(load_chunk)(ir, collection, query, projection) for ir in id_ranges ]
      df = dd.from_delayed(dfs)
      return df
  
  collection = connect_to_mongo_and_return_collection_object(credentials)
  
  # df = parallel_load(collection=collection)
  
  id_ranges = compute_id_ranges(collection)
  dedf = delayed(load_chunk)(id_ranges[0], collection)
  </code></pre>
  
  <p><code>load_chunk</code> perfectly runs when called directly. However, call <code>delayed(load_chunk)( blah-blah-blah )</code> fails with exception, mentioned above.</p>