BlogCategoryTag

Header

Footer

PreFooter

ProseBlock

Blog

SocialLinks

Prose

AnnouncementContextProvider

UnnamedGlobalGroupOfVariants2

Screen

Segment

Experiment

Experiment2

scale.com (Production)

plasmic-nav

plasmic-embed-css

plasmic-basic-components

plasmic-tabs

plasmic-cms

react-chartjs-2

react-slick

react-quill

Chiao-Lun Cheng

Engineering

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">

How to make MongoDB not suck for analytics - Scale

How to make MongoDB not suck for analytics

open-dataset-nucleus

slugs must be all lowercase. ie: ‘my-slug’

events-banner

This entry represents both interview quote highlight and interview chat highlight because of Builder limitations. Choose the fields that matches either of them.

interview-highlight

If it's selected, the author and quote fields will be taken, otherwise the chat fields. The default is for chat.

resources-download-model

transform-session

CTA to open the session. Leave blank if you don't want a cta for that session

Link for the CTA. Usually the link to the exchange session

blog-filter-type

event-transform-x-ticket

solution

homepage-quotes

logos

homepage-content

builder-debug

transfrom-x-in-person

event-model

Toggle if the event is featured. Featured events will be moved to the top of the list

guides-cta

summit-2023-partners

sales

z[deprecated] page

quotes

z[deprecated] compare

product

blog-category-model

announcement-banner

accelerate

Vercel Powered and Edge Computed OG Image. Preferred over og img file upload. If this field is non empty, it will take precedent over the uploaded file

summit-2023-speakers

homepage-resources

customers

Display order on the page /customers. Higher values show first

page-cross-product-webinar

interview

event-transform

Event model for new marketing events page

event-2-0

homepage-event

custom-fonts

transform-x-speaker

Be careful to not add any trailing white spaces to the slug. Slugs must be all lowercase. ie: ‘my-slug’

Order of the featured speakers displayed on the home page for transformX

z[deprecated] document-ai

summit-2023-agenda

event-converge

hosts

author

transform-x

Page: /In-Person. Section: Hero. Text: Title Text

Page: /In-Person. Section: Slider Card. Text: Title Text

Page: /In-Person. Section: Slider Card. Text: Card Body

Page: /In-Person. Section: Hero. Text: Tagline

open-dataset

z[deprecated] customers

URL redirections for Website. Disclaimer: takes 5~ min for each added or edited.

url-redirects

Use relative path. use `/blog/foo` not `https://scale.com/blog/foo`

legal

event

Optional second image intended to show on the grid view, should be a more squared image closer to 1:1 aspect ratio than the one used in the Cover field.

guides

z[deprecated] guides-old

z[deprecated] guides-lp

partners

blog-model

Option to list or not list this post in the Blog index. Enabled = not list in the index, Disabled = list in the index

Products

Government

Solutions

Resources

Customers

Pricing →

How to make MongoDB not suck for analytics

The future of your industry starts here.