Chiao-Lun Cheng

Deep Learning models have amazing capacity, <a href="https://ai.googleblog.com/2017/07/revisiting-unreasonable-effectiveness.html" rel="noopener noreferrer" target="_blank">getting better with more data</a> seemingly without limit. To get a well-functioning model, it is not enough to have large amounts of data, you also need accurate annotations. Although large amounts of data will help the model resolve data inconsistencies between different annotations, humans can still make repeated mistakes that become ingrained in the model. For example, when asked to draw a box around an object, it is natural for humans to always ensure that the box contains the object - i.e., to err on the side of making the box too big. Using such a model for collision avoidance will result in false positives, causing the autonomous vehicle to halt unnecessarily. The oversizing of bounding boxes is an example of a systematic error, as opposed to a random error. Random and systematic errors impact the trained model differently. <h2>Random vs Systematic Errors</h2> Random errors cause the model to require more data to converge to the same result. A model trained with data with random errors will ultimately arrive at the same parameters, just more slowly and after having been fed more data. Systematic errors cause the model to converge to a different result - this does not get better with data quantity. A model trained with oversized boxes will output oversized boxes regardless of how much data it is trained with. <h2>Mitigation Strategies</h2> One simple, if expensive, way to deal with random errors is to use more training data. There are also other more sophisticated and cost-efficient approaches - for example, if your training data has multiple sources, and if each source has a different amount of random errors, you can mitigate the effects of the errors by weighing the low-error sources more heavily while training. Systematic errors, on the other hand, can only be resolved by fixing the data, or understanding the nature of error well enough to turn it into a random error. Many interesting techniques exist in this area, including <a href="https://arxiv.org/abs/1802.05300" rel="noopener noreferrer" target="_blank">Golden Label Correction</a>, where you carefully annotate a subset of the data in order to characterize the non-randomness of the rest of the data. At the end of the day, the key to overcoming annotation errors effectively and efficiently is understanding how the errors are generated. Let’s take the previous example of oversized bounding boxes. Taking into account the tendency of inexperienced workers to oversize boxes, Scale is able to use simple models to check for oversizing. Over time, with feedback from those models, workers learn to make fewer systematic mistakes, consistently drawing tight bounding boxes. By matching complex annotation specifications with annotator training and an understanding of the systematic errors, Scale is able to efficiently produce quality annotations in quantity, facilitating cost-effective training of accurate models.

Training Data - Quantity vs Quality of Annotated Data - Scale

Training Data - Quantity is no Panacea

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

Chiao-Lun Cheng

Training Data - Quantity is no Panacea

How to make MongoDB not suck for analytics