TLDR: Row stores are fast to write but slow to read. Column stores are fast to read but slow to write. Load data from Mongo into Parquet files for fast querying using AWS Athena. MongoDB is a schema-less NoSQL document store that uses a JSON-like format for each document. Each document in Mongo WiredTiger is stored as a contiguous binary blob, which makes our MongoDB instance a row store. AWS Athena is a SaaS offering by Amazon that queries files on S3 using Presto, a distributed SQL query engine that can query high-performance columnar formats like Parquet and ORC. Here at Scale we use MongoDB backed by WiredTiger as our primary data store. This blog post describes why and how we load data from MongoDB into AWS Athena using Parquet files in order to get fast ad-hoc analytics queries on our data.<h2>Row stores and Column stores</h2> Suppose you want to get the sum of a single column(field) over all the rows(documents) in a table(collection). With a Row Store, without an index, to get a field you need to fetch the document, and to get the same field for all documents you would have to essentially load the entire table from disk. All the columns in your table (even columns that do not appear in your query) get fetched as well, which is slow:<img src="https://cdn.builder.io/api/v1/image/assets%2Fe0438815ba51486bbb6a202747122d4b%2Ff746257aeb8d477d9d6d912f1a328ce7" alt="Data Fields"> You can speed up this query by building a covering index. However, each index you add increases disk usage and slows down inserts. Also, when doing data exploration it can be difficult to predict what indices you need ahead of time. With a Column Store, the query is much more efficient. A column store puts data from the same column together. This would allow us to just retrieve the columns we are filtering / aggregating over when doing queries. This allows us to get good performance on ad-hoc queries, without indices:<img src="https://cdn.builder.io/api/v1/image/assets%2Fe0438815ba51486bbb6a202747122d4b%2Fae046f529b8a44e79aa710d117685d41" alt="Column Fields"> Column stores, however, make it very difficult to update a single record - due to the way data is arranged you have to move all of the data to change one record. Also, column stores require us to commit to a schema, to define the columns ahead of time.<h2> </h2><h2> Trading off read and write performance</h2> The slowness of column-store writes is related to the size of the table, because you need to rewrite the whole table to change a single record. By representing each row-store table as multiple column-store tables, we can get the column store to have faster writes at the expense of slowing down its reads. For our implementation, this means we map each Mongo collection to multiple Parquet files (each file actually contains multiple row groups, with each row group being column-major, but we use files as a unit for convenience), and refresh all the documents in a Parquet file whenever any one of them changes. We use smaller file sizes than what would be optimal for reads, in order to make writes faster and reduce replication lag.<h2>Cache Invalidation</h2> When updating the column store from reading the row store, we need to <ol><li>Figure out which records are stale</li><li>Update those records efficiently</li><li> </li></ol> To determine staleness, we need to know the identity of the document and the time of modification. This data is obtained by reading the Mongo oplog, a data structure that represents all inserts/updates/deletes to the database chronologically, available on Mongo replica sets for internal use in replication from primary to secondary nodes within the cluster. For efficiency, we want to ensure that documents to be modified belong in as few files as possible. We achieve this by grouping records by their Mongo <a href="https://docs.mongodb.com/manual/reference/method/ObjectId/" rel="noopener noreferrer" target="_blank">ObjectIDs</a>. We are making use of the fact that, like in <a href="http://wiki.c2.com/?GenerationalGarbageCollection" rel="noopener noreferrer" target="_blank">Generational Garbage Collection</a>, young documents tend to get modified more than old ones. We maintain an index recording the ObjectID ranges of each file, starting a new file whenever the newest bucket is filled with more than N documents.<pre class="ql-syntax" spellcheck="false">class Buckets(object):
 def __init__(self, size_limit):
 self.size_limit = size_limit
 self.buckets = [ObjectId("0" * 24)]
 self.max_bucket = set()

 def find_le(self, x):
 'Find rightmost value less than x'
 i = bisect.bisect_left(self.buckets, x)
 if i:
 return i - 1
 raise ValueError

 def get_bucket(self, oid):
 i = self.find_le(oid)

 # oid is in one of lower buckets
 if i &amp;lt; len(self.buckets) - 1:
 return self.buckets[i], self.buckets[i + 1]

 # oid is in maximum bucket
 self.max_bucket.add(oid)

 if len(self.max_bucket) &amp;lt; self.size_limit:
 return self.buckets[i], max(self.max_bucket)

 # maximum bucket is oversized, start a new bucket
 self.buckets.append(max(self.max_bucket))
 self.max_bucket = set()
 return self.buckets[i], self.buckets[i + 1]
</pre><h2>Implementation Details</h2> <ol><li> </li><li> The scripts that read from mongo and create parquet files are written in</li><li> Python and use the</li><li> <a href="https://pypi.org/project/pyarrow" rel="noopener noreferrer" target="_blank">pyarrow library</a> to write</li><li> Parquet files. The first version implemented a filter-and-append strategy</li><li> for updating Parquet files, which works faster than overwriting the entire</li><li> file. This proved to be a mistake as pyarrow (8.9.0) support for reading is</li><li> much less mature than that for writing. We would occasionally lose data when</li><li> reading rows for rewriting into the new parquet file, and eventually gave up</li><li> and went to an implementation that never read from the Parquet files.</li><li> </li><li> </li><li> Each Parquet file is made of row groups, and there is a size limit for each</li><li> row group. We batch documents by size to avoid this problem.</li><li> </li><li> </li></ol><pre class="ql-syntax" spellcheck="false">def islice_docs(docs, size):
 tot = 0
 for doc in docs:
 tot += len(doc.raw)
 yield doc
 if tot &gt; size * 1024**2:
 return

def batch_docs(docs, size):
 sourceiter = iter(docs)
 while True:
 batchiter = islice_docs(sourceiter, size)
 try:
 yield itertools.chain([next(batchiter)], batchiter)
 except StopIteration:
 return
</pre><ol><li> </li><li> When using pymongo, use RawBSONDocument to defer parsing. In the case of</li><li> multiple processes writing different Parquet files, this means you avoid</li><li> having parsing happen in the serial part of your program. You can</li><li> selectively parse sub-documents using bson.BSON</li><li> </li><li> </li></ol><pre class="ql-syntax" spellcheck="false"> def realize_raw_bson(d):
 if isinstance(d, RawBSONDocument):
 return bson.BSON(d.raw).decode()
 if isinstance(d, list):
 return [realize_raw_bson(v) for v in d]
 return d
</pre><ol><li> </li><li> Use <a href="https://pypi.org/project/atomicwrites/" rel="noopener noreferrer" target="_blank">atomicwrites</a> to</li><li> make writing a new Parquet file all-or-nothing</li><li> </li><li> </li></ol><pre class="ql-syntax" spellcheck="false">@contextmanager
 def atomic_parquet_writer(output, *args):
 os.makedirs(os.path.dirname(output), exist_ok=True)
 with atomic_write(output, mode="wb", overwrite=True) as f:
 with pq.ParquetWriter(f, *args) as writer:
 yield writer
</pre><ol><li> </li><li> On top of extracting explicitly enumerated fields, put the whole json in a</li><li> column by itself for querying fields not in the schema</li><li> </li><li> </li><li> For reading the Parquet files locally using SQL,</li><li> <a href="https://drill.apache.org/" rel="noopener noreferrer" target="_blank">Apache Drill</a> is very convenient -</li><li> just untar and run drill-embedded</li><li> </li><li> </li><li> After updating a parquet file, it has to be copied to S3 (using sync, not</li><li> cp). Using EC2 is essential here because we are constantly copying files</li><li> from the server to S3, and the bandwidth costs would be prohibitive</li><li> otherwise - traffic between EC2 and S3 in the same region is free. After</li><li> copying the file to S3, use AWS Glue to discover the schema from the files -</li><li> call Glue only when the schema changes, AWS charges you each time you call</li><li> it.</li><li> </li><li> </li><li> The oplog contains all the information you need to update the Parquet files.</li><li> We run an instance of mongo on the same machine, and maintain a one-node</li><li> slave as it were by using the applyOps command on oplog entries read from</li><li> the main cluster. Beware, however - Mongo developers might</li><li> <a href="https://jira.mongodb.org/browse/SERVER-33182" rel="noopener noreferrer" target="_blank">remove the docs</a></li><li> for applyOps completely in the future.</li><li> </li><li> </li></ol><img src="https://cdn.builder.io/api/v1/image/assets%2Fe0438815ba51486bbb6a202747122d4b%2Fdf9dcbefbd7f43ce97d5b6780e60ac25" alt="Scale's flow">