title: Best practice to query large number of ndb entities from datastore
body: <p>I have run into an interesting limit with the App Engine datastore.  I am creating a handler to help us analyze some usage data on one of our production servers.  To perform the analysis I need to query and summarize 10,000+ entities pulled from the datastore.  The calculation isn't hard, it is just a histogram of items that pass a specific filter of the usage samples.  The problem I hit is that I can't get the data back from the datastore fast enough to do any processing before hitting the query deadline.</p>
  
  <p>I have tried everything I can think of to chunk the query into parallel RPC calls to improve performance, but according to appstats I can't seem to get the queries to actually execute in parallel. No matter what method I try (see below) it always seems that the RPC's fall back to a waterfall of sequential next queries.</p>
  
  <p>Note: the query and analysis code does work, it just runs to slowly because I can't get data quickly enough from the datastore.</p>
  
  <h1>Background</h1>
  
  <p>I don't have a live version I can share, but here is the basic model for the part of the system I am talking about:</p>
  
  <pre><code>class Session(ndb.Model):
     """ A tracked user session. (customer account (company), version, OS, etc) """
     data = ndb.JsonProperty(required = False, indexed = False)
  
  class Sample(ndb.Model):
     name      = ndb.StringProperty  (required = True,  indexed = True)
     session   = ndb.KeyProperty     (required = True,  kind = Session)
     timestamp = ndb.DateTimeProperty(required = True,  indexed = True)
     tags      = ndb.StringProperty  (repeated = True,  indexed = True)
  </code></pre>
  
  <p>You can think of the samples as times when a user makes use of a capability of a given name.  (ex: 'systemA.feature_x').  The tags are based upon customer details, system information, and the feature.  ex: ['winxp', '2.5.1', 'systemA', 'feature_x', 'premium_account']).  So the tags form a denormalized set of tokens that could be used to find samples of interest.</p>
  
  <p>The analysis I am trying to do consists of taking a date range and asking how many times was a feature of set of features (perhaps all features) used per day (or per hour) per customer account (company, not per user).</p>
  
  <p>So the input to the handler be something like:</p>
  
  <ul>
  <li>Start Date</li>
  <li>End Date</li>
  <li>Tag(s)</li>
  </ul>
  
  <p>Output would be:</p>
  
  <pre><code>[{
     'company_account': &lt;string&gt;,
     'counts': [
        {'timeperiod': &lt;iso8601 date&gt;, 'count': &lt;int&gt;}, ...
     ]
   }, ...
  ]
  </code></pre>
  
  <h1>Common Code for Queries</h1>
  
  <p>Here is some code in common for all queries.  The general structure of the handler is a simple get handler using webapp2 that sets up the query parameters, runs the query, processes the results, creates data to return.</p>
  
  <pre><code># -- Build Query Object --- #
  query_opts = {}
  query_opts['batch_size'] = 500   # Bring in large groups of entities
  
  q = Sample.query()
  q = q.order(Sample.timestamp)
  
  # Tags
  tag_args = [(Sample.tags == t) for t in tags]
  q = q.filter(ndb.query.AND(*tag_args))
  
  def handle_sample(sample):
     session_obj = sample.session.get()    # Usually found in local or memcache thanks to ndb
     count_key   = session_obj.data['customer']
     addCountForPeriod(count_key, sample.timestamp)
  </code></pre>
  
  <h1>Methods Tried</h1>
  
  <p>I have tried a variety of methods to try to pull data from the datastore as quickly as possible and in parallel.  The methods I have tried so far include:</p>
  
  <h2>A. Single Iteration</h2>
  
  <p>This is more of a simple base case to compare against the other methods.  I just build the query and iterate over all the items letting ndb do what it does to pull them one after the other.</p>
  
  <pre><code>q = q.filter(Sample.timestamp &gt;= start_time)
  q = q.filter(Sample.timestamp &lt;= end_time)
  q_iter = q.iter(**query_opts)
  
  for sample in q_iter:
     handle_sample(sample)
  </code></pre>
  
  <h2>B. Large Fetch</h2>
  
  <p>The idea here was to see if I could do a single very large fetch.</p>
  
  <pre><code>q = q.filter(Sample.timestamp &gt;= start_time)
  q = q.filter(Sample.timestamp &lt;= end_time)
  samples = q.fetch(20000, **query_opts)
  
  for sample in samples:
     handle_sample(sample)
  </code></pre>
  
  <h2>C. Async fetches across time range</h2>
  
  <p>The idea here is to recognize that the samples are fairly well spaced across time so I can create a set of independent queries that split the overall time region into chunks and try to run each of these in parallel using async:</p>
  
  <pre><code># split up timestamp space into 20 equal parts and async query each of them
  ts_delta       = (end_time - start_time) / 20
  cur_start_time = start_time
  q_futures = []
  
  for x in range(ts_intervals):
     cur_end_time = (cur_start_time + ts_delta)
     if x == (ts_intervals-1):    # Last one has to cover full range
        cur_end_time = end_time
  
     f = q.filter(Sample.timestamp &gt;= cur_start_time,
                  Sample.timestamp &lt; cur_end_time).fetch_async(limit=None, **query_opts)
     q_futures.append(f)
     cur_start_time = cur_end_time
  
  # Now loop through and collect results
  for f in q_futures:
     samples = f.get_result()
     for sample in samples:
        handle_sample(sample)
  </code></pre>
  
  <h2>D. Async mapping</h2>
  
  <p>I tried this method because the documentation made it sound like ndb may exploit some parallelism automatically when using the Query.map_async method.</p>
  
  <pre><code>q = q.filter(Sample.timestamp &gt;= start_time)
  q = q.filter(Sample.timestamp &lt;= end_time)
  
  @ndb.tasklet
  def process_sample(sample):
     period_ts   = getPeriodTimestamp(sample.timestamp)
     session_obj = yield sample.session.get_async()    # Lookup the session object from cache
     count_key   = session_obj.data['customer']
     addCountForPeriod(count_key, sample.timestamp)
     raise ndb.Return(None)
  
  q_future = q.map_async(process_sample, **query_opts)
  res = q_future.get_result()
  </code></pre>
  
  <h1>Outcome</h1>
  
  <p>I tested out one example query to collect overall response time and appstats traces.  The results are:</p>
  
  <h2>A. Single Iteration</h2>
  
  <p>real: 15.645s</p>
  
  <p>This one goes sequentially through fetching batches one after the other and then retrieves every session from memcache.</p>
  
  <p><img src="https://i.stack.imgur.com/CSyR1.png" alt="Method A appstats"></p>
  
  <h2>B. Large Fetch</h2>
  
  <p>real: 12.12s</p>
  
  <p>Effectively the same as option A but a bit faster for some reason.</p>
  
  <p><img src="https://i.stack.imgur.com/HsPZR.png" alt="Method B appstats"></p>
  
  <h2>C. Async fetches across time range</h2>
  
  <p>real: 15.251s</p>
  
  <p>Appears to provide more parallelism at the start but seems to get slowed down by a sequence of calls to next during iteration of the results.  Also doesn't seem to be able to overlap the session memcache lookups with the pending queries.</p>
  
  <p><img src="https://i.stack.imgur.com/Ty2l4.png" alt="Method C appstats"></p>
  
  <h2>D. Async mapping</h2>
  
  <p>real: 13.752s</p>
  
  <p>This one is the hardest for me to understand.  It looks like it has q good deal of overlapping, but everything seems to stretch out in a waterfall instead of in parallel.</p>
  
  <p><img src="https://i.stack.imgur.com/JoNlN.png" alt="Method D appstats"></p>
  
  <h1>Recommendations</h1>
  
  <p>Based upon all this, what am I missing?  Am I just hitting a limit on App Engine or is there a better way to pull down large number of entities in parallel?  </p>
  
  <p>I am at a loss as to what to try next.  I thought about rewriting the client to make multiple requests to app engine in parallel but this seems pretty brute force.  I would really expect that app engine should be able to handle this use case so I am guessing there is something I am missing.</p>
  
  <h1>Update</h1>
  
  <p>In the end I found that option C was the best for my case. I was able to optimize it to complete in 6.1 seconds.  Still not perfect, but much better.</p>
  
  <p>After getting advice from several people, I found that the following items were key to understand and keep in mind:</p>
  
  <ul>
  <li>Multiple queries can run in parallel</li>
  <li>Only 10 RPC's can be in flight at once</li>
  <li>Try to denormalize to the point that there are no secondary queries</li>
  <li>This type of task is better left to map reduce and task queues, not real-time queries</li>
  </ul>
  
  <p>So what I did to make it faster:</p>
  
  <ul>
  <li>I partitioned the query space from the beginning based upon time.  (note: the more equal the partitions are in terms of entities returned, the better)</li>
  <li>I denormalized the data further to remove the need for the secondary session query</li>
  <li>I made use of ndb async operations and wait_any() to overlap the queries with the processing</li>
  </ul>
  
  <p>I am still not getting the performance I would expect or like, but it is workable for now.  I just wish their was a better way to pull large numbers of sequential entities into memory quickly in handlers.</p>