How to improve your MongoDB write throughput with SQS
Tags: aws, sqs, mongodb, npm
Published: 2021-11-15 14:11:47

---
title: How to improve your MongoDB write throughput with SQS
published: true
description: Deep dive into how we scaled our MongoDB write througput by queueing up operations with AWS SQS and then processing those to bulk operations
tags: aws, sqs, mongodb, npm,
cover_image: https://purpleads.io/blog/wp-content/uploads/2021/11/mongo-sqs-825x510.png
---

<h2><strong>What is SQS?</strong></h2>
<span style="font-weight: 400;">Amazon Simple Queue (SQS) is a highly scalable managed message queue service. It lets you decouple different parts of your application and eliminate throughput bottlenecks, for example high frequency write operations from your application to your database (like in our case with MongoDB). </span><span style="font-weight: 400;">SQS really is a simple service to use with just a few API functions and you can get up and running quickly.</span><span style="font-weight: 400;">
</span><span style="font-weight: 400;">It’s important to note SQS is </span><a href="https://aws.amazon.com/sqs/pricing/"><span style="font-weight: 400;">billed by the number of requests</span></a><span style="font-weight: 400;"> and that you would need to optimize at high scale.</span>
<h2><strong>Why We Started Using SQS</strong></h2>
<span style="font-weight: 400;">At <a href="https://purpleads.io">PurpleAds</a> we work mainly with AWS EC2. It was easier for us to start with, it keeps costs low compared to managed services and you get more control over your infrastructure albeit at the cost of rising complexity as you grow.
</span><span style="font-weight: 400;">Since we like to keep things simple and use technologies we are familiar with - we use a 2x <code>a1.xlarge</code> mongodb replica set for all our database needs. </span>

<span style="font-weight: 400;">Running an ad network requires storing a lot of analytics and recently, our mongo servers were having a hard time keeping up with increasing numbers of write and update operations as our network grows.
</span><span style="font-weight: 400;">After some research, looking a lot at outputs of </span><i><span style="font-weight: 400;">mongostat </span></i><span style="font-weight: 400;">and </span><i><span style="font-weight: 400;">mongotop</span></i><span style="font-weight: 400;">, it seems that mongo is not so great when it comes to updating hundreds of fields in hundreds of documents per minute. Imagine a daily analytics document keeping data per asset, per device, per geo, per os, etc.
</span>

<img class="wp-image-530 size-large" src="https://purpleads.io/blog/wp-content/uploads/2021/11/Screen-Shot-2021-11-15-at-15.18.45-1024x366.png" alt="mongodb overloaded" width="1024" height="366" /> The MongoDB CPU Average, at peak times it was at 97% average and crashed daily

<span style="font-weight: 400;">So we decided to queue these ops and do periodical bulk writes instead, this would cut down the number of upsert operations by at least 10x.
</span><span style="font-weight: 400;">Several solutions came to mind, among others we though about queueing operations in our Node application memory.</span><span style="font-weight: 400;">
</span><span style="font-weight: 400;">Finally we decided to go with something that we can implement quickly, easily and will give us peace of mind for a long period - AWS Simple Queue Service.</span>

<img class="wp-image-532 size-large" src="https://purpleads.io/blog/wp-content/uploads/2021/11/Screen-Shot-2021-11-15-at-15.19.57-1024x342.png" alt="mongodb with sqs" width="1024" height="342" /> MongoDB CPU Average with SQS queued bulk writes, never crosses 50% average
<span style="font-weight: 400;">Having your main Node app write thousands of upsert operations to MongoDB not only overloads the database and creates locks, it also makes your Node app less responsive as it has a lot of open connections to your database.</span>

<span style="font-weight: 400;">This is not going to be one of those generic articles about SQS, we’re going to dive into how we used it to improve our mongo write performance.
The main idea was simple - send payloads to SQS and process it later in bulk with a separate worker application (a microservice if you will).</span>

Since SQS pricing model is based on the number of requests you make. A message has to be sent, read and deleted - a total of 3 requests per message batch.
A single request can contain up to 10 messages/payloads and a maximum of 256 KB payload. So we're going to try to <span style="font-weight: 400;">leep costs to minimum and </span><span style="font-weight: 400;">offload the heavy lifting from our main app and database.</span>
<h2><strong>Sending Messages</strong></h2>
<span style="font-weight: 400;">The first and simple cost and performance optimization on SQS is queueing message sends.
We can send 10 messages per request, so we'll stack them up before sending them over to SQS.
You should also setup a scheduler to send messages periodically if you don’t reach the 10 message threshold in a given time. This will help you avoid losing messages on process interruptions (e.g. auto scale events, crashing, etc)</span>

<span style="font-weight: 400;">Here’s how we went about handling that:</span>
<div class="hcb_wrap">
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">const messages = [];
// setup a simple scheduler
setInterval(() =&gt; {
  _sendQueue();
}, 10 * 1000);

// use this when you want to send a JSON message
function queueSend(object) {
    messages.push(object);
    if(messages.length &gt;= 10) {
      _sendQueue();
    }
}

// this should not be called directly
function _sendQueue() {
  if(!this.messages || !this.messages.length) {
    return;
  }
  // max batch size is 10
  const toProcess = messages.splice(0, 10);
  let i = 0;
  const entries = toProcess.map((mes) =&gt; {
    i += 1;
    return { Id: `${i}`, MessageBody: JSON.stringify(mes) };
  });
  sqs.sendMessageBatch({ QueueUrl, Entries: entries }).promise();
}</pre>
</div>
<span style="font-weight: 400;">In case your payloads are small, you can really bring down your costs by “fooling” SQS and sending an array of payloads in a single message.</span>

<span style="font-weight: 400;">Our JSON payload was around 500 bytes on average of a JSON string. Basically we could send around 100k payloads in one message, sending 1000x less requests/messages. You would need to write a more complex processing algorithm that extracts the multiple payloads from a single message when pulling those, but that’s for another article.</span>
<h2><strong>Pulling Messages</strong></h2>
<span style="font-weight: 400;">Now that you have your messages in your SQS queue, you need a worker, a micro-app that would process those messages and write them into MongoDB in bulk.
</span><span style="font-weight: 400;">SQS allows to pull up to 10 messages from the queue at a time (it can be less or even zero, more on that later), MongoDB on the other hand can send up to 1000 operations in a single bulk transaction.
</span><span style="font-weight: 400;">A good approach here would be:</span>
<ol>
 	<li style="font-weight: 400;" aria-level="1"><span style="font-weight: 400;">concurrently pull 1000 or more messages in batches of 10</span></li>
 	<li style="font-weight: 400;" aria-level="1"><span style="font-weight: 400;">process / transform / map those messages to mongodb update/upsert operations</span></li>
 	<li style="font-weight: 400;" aria-level="1"><span style="font-weight: 400;">bulk write these operations to mongodb</span></li>
 	<li style="font-weight: 400;" aria-level="1"><span style="font-weight: 400;">delete the messages from the SQS queue</span></li>
</ol>
<span style="font-weight: 400;">Here’s how you can go about pulling those messages:</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">function sqsPull() {
  return sqs.receiveMessage({
    QueueUrl,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
  }).promise();
}

const multiPulls = await Promise.all(
  Array(100).fill(0).map(() =&gt; sqsPull())
);
const messages = multiPulls.map((res) =&gt; res.Messages).flat().filter((i) =&gt; !!i) || [];
// you need these to later delete the messages from the queue
const receiptIds = messages.map((m) =&gt; m.ReceiptHandle);
const json = messages.map((m) =&gt; {
  try { 
    return JSON.parse(m.Body)
  } catch(err) {
    return null;
  }
});
</pre>
<span style="font-weight: 400;">Pulling messages is nice, but what if our queue is empty? Or if we get a lot of small batches with less than 10 messages? We are paying for those requests but not maximizing our cost effectiveness.
</span><span style="font-weight: 400;">It’s important to note that sometimes SQS will return less than 10 messages per request even if you have hundreds of messages in-flight, probably these are still being processed by SQS and not ready to be pulled.</span>

<img class="wp-image-534 size-large" src="https://purpleads.io/blog/wp-content/uploads/2021/11/Screen-Shot-2021-11-15-at-15.26.00-1024x408.png" alt="sqs empty receives" width="1024" height="408" /> Make sure to monitor your SQS empty receives section to avoid incurring costs for empty requests

<span style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen-Sans, Ubuntu, Cantarell, 'Helvetica Neue', sans-serif;">That’s why we’ve setup a “cooldown” mechanism to let the queue stack up on more messages before we process those once again. </span><span style="font-weight: 400;">If we got 10 consecutive batches of less than 1,000 messages, we let the worker sleep for a while.</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">let smallBatchCounter = 0;
do {
  const multiPulls = await Promise.all(Array(100).fill(0).map(() =&gt; sqsPull()));
  const messages = multiPulls.map((res) =&gt; res.Messages).flat().filter((i) =&gt; !!i) || [];
  ...
  if(message.length &lt; 100 * 1000) {
    smallBatchCounter += 1;
  } else {
    smallBatchCounter = 0;
  }
  if(smallBatchCounter &gt;= 10) {
    await new Promise((resolve) =&gt; setTimeout(resolve, 5 *60 * 1000));
  }
} while(true);</pre>
<h2><strong><span style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen-Sans, Ubuntu, Cantarell, 'Helvetica Neue', sans-serif;">Processing Messages</span></strong></h2>
<span style="font-weight: 400;">Processing messages is very specific to your application and schema and use case but we’ll show you what we did anyway so you can get ideas for your application.</span>

<span style="font-weight: 400;">In our case, since we’re doing analytics, these are usually many field increments on a single document. So a single document could be updated by many messages in one batch (e.g. 5 requests related to one account would be 5 times { $inc: 1 }), we want to have one update operation per document.</span>

<span style="font-weight: 400;">An example of how a message looks like in our case:</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">{
  "accountId": string,
  "assetId": string,
  "date": timestamp,
  "event": "click",
  "cpc": 0.5
}
</pre>
<span style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen-Sans, Ubuntu, Cantarell, 'Helvetica Neue', sans-serif;">You can see that there could be multiple events for a combination of <em>accountId</em>, <em>assetId</em>, <em>date</em>.
</span><span style="font-weight: 400;">So we would map these out like this:</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">{
  "accountId_assetId_date": {
    "$inc": { "clicks": 1, ​"revenue": 0.5 },
  },
  ....
}</pre>
<span style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen-Sans, Ubuntu, Cantarell, 'Helvetica Neue', sans-serif;">Now for every 1000 messages we would get far less operations to write to mongo.
</span><span style="font-weight: 400;">After we have our map, we would create an array of operations to bulkWrite.</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">const entries = Object.entries(map);
// note how we increment by 1000 every iteration (the max number of operations for mongo per bulkWrite)
for(let i = 0; i &lt; entries.length; i += 1000) {
  const ops = entires.slice(i, i + 1000).map(([key, value]) =&gt; {
    const [accountId, assetId, date] = key.split('_');
    const filterQuery = { accountId, assetId, date };
    updateOne: {
      filter: filterQuery,
      update: value,
      upsert: true,
    },
  });
  await model.bulkWrite(ops, { ordered: false });
}</pre>
<h2><strong>Deleting Messages</strong></h2>
<span style="font-weight: 400;">You would not want your messages to be processed more than once and save duplicate data. So after successfully processing messages, you should delete those.
</span><span style="font-weight: 400;">Here again, SQS limits you to deleting up to 10 messages in a single request. So we added a helper to delete larger batches:</span>
<pre class="EnlighterJSRAW" data-enlighter-language="js" data-enlighter-theme="beyond">function deleteBatch(receiptIds = []) {
  if(receiptIds.length &gt; 10) {
    throw new Error('max 10 ids per delete batch');
  }
  let i = 0;
  const entries = receiptIds.map((rId) =&gt; {
    i += 1;
    return { Id: `${i}`, ReceiptHandle: rId };
  });
  return sqs.deleteMessageBatch({ QueueUrl, Entries: entries }).promise();
}

function deleteLargeBatch(receiptIds = []) {
  const promises = [];
  for(let i = 0; i &lt; receiptIds.length; i += 10) {
    promises.push(deleteBatch(receiptIds.slice(i, i + 10)));
  }
  return Promise.all(promises);
}</pre>
<h2><strong><span style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen-Sans, Ubuntu, Cantarell, 'Helvetica Neue', sans-serif;">Why not FIFO queue?</span></strong></h2>
<span style="font-weight: 400;">First In First Out SQS queues guarantee that messages will be pulled in the order they were sent and will only be delivered once to the consumer/client.
</span><span style="font-weight: 400;">The caveat with using FIFO queues are MessageGroupId. A message has to be associated with a group and the order of messages is only guaranteed relative to the group.
</span><span style="font-weight: 400;">A single message group can have up to 20k messages in-flight or “waiting” to be pulled.
</span><span style="font-weight: 400;">What if we have more than 20k? We would have to split those up to multiple groups, but then what would be the order of messages relative to other groups? We could not find an answer to this.</span>

<span style="font-weight: 400;">For our use case simple queues were sufficient, messages are delivered in-order and no messages are left behind.</span>
<h2><strong>Conclusion</strong></h2>
<span style="font-weight: 400;">All-in-all AWS SQS is a cheap, simple, effective and scalable service. If you’re having trouble with write through-puts, you should consider using SQS to queue and batch these operations.
</span><span style="font-weight: 400;">Hopefully this guide will help you get started and later on help you optimize your costs and efficiency.</span>

<span style="font-weight: 400;">We’ve also published our </span><a href="https://www.npmjs.com/package/@purpleads/sqs-client"><span style="font-weight: 400;">SQS helper library on npm</span></a><span style="font-weight: 400;"> with many of the operations described in this article for your convenience.</span>