Serverless and (Micro)Batch Processing: AWS Lambda and the S3 data lake

At Warren Rogers we are actively working to modernize our data processing pipelines by leveraging AWS managed services. Although many processes that were once executed in batch are being done in our new real time complex event processing pipeline using Amazon Kinesis, it is still practical (and more affordable) for some processes to run in batch at regular intervals (once a day or month for example). In the legacy system, jobs against file-based data can run for hours, basically using up the valuable resources of a single large server. We needed an efficient and practical way to borrow compute power for a period of time. We soon realized that serverless functions with AWS Lambda are the perfect fit for many of these jobs for a few reasons.

Firstly, our data lake, like in many other companies is stored in Amazon S3. This turned out to be a great decision. Not only is S3 scalable and affordable, but Amazon recently released an incredibly useful optimization tool called S3 Select. Using simple SQL queries, we can pull the exact fields from the exact rows we need for our batch processes. Running S3 Select over large amounts of data (effectively using server-side filtering) can take a relatively short amount of time, which is important when running AWS Lambda functions.

Secondly, running a serverless function against S3 is perfect as Lambdas can run using hundreds of concurrent executions while S3 was built to handle massive scale. Running Lambda unfettered against something like a traditional database would be sure to take it down. In general, one needs to be very careful to take note of the resources a highly scalable and concurrent service like Lambda calls upon. In the best case your Lambdas are running at full speed, with all the allowed concurrency, against a highly scalable backend like S3 or DynamoDB. Otherwise there are ways to reduce the concurrency and hamstring Lambda if you have to (see reserved concurrency).

Thirdly, our data and its processing are naturally segmented by site. This means we can cut up processing into thousands of jobs in a very natural way. Running concurrent Lambdas against S3 has some of the same constraints you might think of when it comes to thread safety as S3 doesn’t make strong consistency guarantees. You don’t want to share data (in particular generated artifacts) across different instances of the job. We have naturally “single-threaded” data that fits nicely with this pattern.

As to the mechanism itself that runs these jobs, we discovered that we could use one Lambda function, triggered by CloudWatch at regular intervals, to grab all the sites (representing discrete jobs) from S3 and place each site as an event in an SNS topic. For each event, we could trigger a Lambda to run that job. I have noticed up to 200 invocations running concurrently using this model.

To use this pattern, there are a few constraints. First, you must be able to break up your job into discrete chunks. Secondly, the job must use no more than 3 GB of memory and run no longer than five minutes. These limits are imposed by AWS Lambda. At Warren Rogers, this fits many of our use cases.

We have found that this pattern uses less resources, costs less and is faster than doing this with say a Hadoop cluster, not to mention all of the headaches of setting up and maintaining virtual machines. Our jobs are less expensive, faster and easier to develop and update using AWS Lambda.

Data Streaming and Storage: Guaranteeing Delivery while Aggregating Data in AWS

At Warren Rogers, we ingest, store and process tens of millions of events from thousands of sites per day. My team was tasked with updating the legacy file-based storage and processing system to a more distributed, highly available and durable streaming solution in AWS.

As we looked at the requirements, it became clear that our data pipeline would need to be a strictly time-ordered stream with guaranteed delivery and very durable persistence. Our data ingestion mechanism would need to be horizontally scalable (and so stateless) and write to a durable storage (so that events could be replayed when needed) as well as publish to a real-time event processing pipeline. An additional requirement was that the data storage solution would need to be efficient at retrieving events for a day or several days for a single IoT sensor (each day holding thousands of events). Also, wherever possible, we would attempt to user AWS managed services so that we could cut on the cost and complexity of maintenance.

It took three attempts to get a solution that satisfied these requirements.

For our first attempt, we targeted various AWS managed services. Amazon S3 became our event storage backend. S3 makes very high durability and availability guarantees and is relatively inexpensive. It also makes long time data storage simpler with lifecycle management. Amazon Kinesis was a natural fit for our real-time processing pipeline and the data ingestion mechanism translated fairly easily to an Amazon Elastic Beanstalk multi-node service.

Time-ordering would be guaranteed on the IoT sensor as it would queue up data and send one event to the data ingestion service at a time. To achieve guaranteed delivery, the data ingestion service would need to place the event in S3 and then in the real-time stream. Only once the event succeeded to post to S3 and the real-time stream did the service return an acknowledgement. If the IoT sensor received an acknowledgement, it would remove the event from the queue and send the next event. Otherwise it would continue to retry the event.

Once we had implemented this system, we came across a problem. It would take tens of seconds (sometimes minutes) to reconstitute one day’s worth of data for a site. Essentially storing each event as an object in S3 made the data storage unusable. Even running a process that would aggregate events after the fact would be expensive. We saw that we needed to group events by IoT sensor (the aggregator Enterprise Integration Pattern) and store them in S3 to get efficient storage.

This highlighted the key problem, how do you aggregate data, maintain data ordering and still guarantee delivery on a multi-node service? The data must be persisted in a durable way before returning an acknowledgement as part of the guaranteed delivery contract. This means it must be stored while it is being aggregated. Obviously writing to disk in a multi-node environment is risky (a step in the wrong direction concerning durability) and would also break ordering as an IoT sensor could deliver the next event to a different node.

Our second rewrite attempted to use EFS (Elastic File System) as a distributed cache where we would group events over some period of time (an hour in site time for example) and then a node would save the file to S3. This allowed for better durability and would maintain ordering.

The problem with EFS is that it was far too slow. There was no way we could keep up with the volume of events that we needed to ingest.

For our third rewrite, we discovered Amazon Data Firehose would take a number of events and aggregate them for us before dumping them to S3. Of course these event objects would need to be grouped and organized by IoT sensor by a continuous background process.

This worked but it seemed highly inefficient to write every record to Firehose. For one, it created an undue resource drain on the data ingestion service. For another, our events were often smaller than 1kb. Firehose rounds up to the next 5kb in cost. In this system, very likely we would be paying five times more on average than we needed. Aggregating before posting to Firehose would be ideal. We  still couldn't aggregate by IoT device and keep our guaranteed delivery contract.

Once the solution dawned on us, it seemed obvious. A service could aggregate across IoT sensors. All the events that happened to be received at a particular service node in some configurable period of time would be grouped and then posted to Firehose. On successful completion, each IoT sensor would receive a confirmation. This had the added benefit of producing a natural throttling mechanism in case of (hopefully very infrequent) downtimes leading to backed up queues.

After the third rewrite, the data ingestion service has been chugging along for a few years now with very few issues.