Chapter 11. Analytics-on-write

In the previous chapter, we implemented a simple analytics-on-read strategy for OOPS, our fictitious package-delivery company, using Amazon Redshift. The focus was on storing our event stream in Redshift in such a way as to support as many analyses as possible “after the fact.” We modeled a fat event table, widened it further with dimension lookups for key entities including drivers and trucks, and then tried out a few analyses on the data in SQL.

For the purposes of this chapter, we will assume that some time has passed at OOPS, during which the BI team has grown comfortable with writing SQL queries against the OOPS event stream as stored in Redshift. Meanwhile, rumblings are coming from various stakeholders at OOPS who want to see analyses that are not well suited to Redshift. For example, they are interested in the following:

In this chapter, we will explore techniques for delivering these kinds of analytics, which broadly fall under the term analytics-on-write. Analytics-on-write has an up-front cost: we have to decide on the analysis we want to perform ahead of time and put this analysis live on our event stream. In return for this constraint, we get some benefits: our queries are low latency, can serve many simultaneous users, and are simple to operate.

To implement analytics-on-write for OOPS, we are going to need a database for storing our aggregates, and a stream processing framework to turn our events into aggregates. For the database, we will use Amazon DynamoDB, which is a hosted, highly scalable key-value store with a relatively simple query API. For the stream processing piece, we are going to try out AWS Lambda, which is an innovative platform for single-event processing, again fully hosted by Amazon Web Services.

Let’s get started!

11.1. Back to OOPS

For this chapter, we will be returning to OOPS, the scene of our analytics-on-read victory, to implement analytics-on-write. Before we can get started, we need to get Kinesis set up and find out exactly what analytics our OOPS bosses are expecting.

11.1.1. Kinesis setup

In the preceding chapter, we interacted with the OOPS unified log through an archive of events stored in Amazon S3. This archive contained the five types of OOPS events, all stored in JSON, as recapped in figure 11.1. This event archive suited our analytics-on-read requirements fine, but for analytics-on-write, we need something much “fresher”: we need access to the OOPS event stream as it flows through Amazon Kinesis in near real-time.

Our OOPS colleagues have not yet given us access to the “firehose” of live events in Kinesis, but they have shared a Python script that can generate valid OOPS events and write them directly to a Kinesis stream for testing. You can find this script in the GitHub repository:

Before we can use this script, we must first set up a new Kinesis stream to write the events to. As in chapter 4, we can do this by using the AWS CLI tools:

That command doesn’t give any output, so let’s follow up by describing the stream:

We are going to need the stream’s ARN later, so let’s add it into an environment variable:

Now let’s try out the event generator:

Great—we can successfully write events to our Kinesis stream. Press Ctrl-C to cancel the generator.

11.1.2. Requirements gathering

With analytics-on-read, our focus was on devising the most flexible way of storing our event stream in Redshift, so that we could support as great a variety of post facto analyses as possible. The structure of our event storage was optimized for future flexibility, rather than being tied to any specific analysis that OOPS might want to perform later.

With analytics-on-write, the opposite is true: we must understand exactly what analysis OOPS wants to see, so that we can build that analysis to run in near real-time on our Kinesis event stream. And, ideally, we would get this right the first time; remember that a Kinesis stream stores only the last 24 hours of events (configurable up to one week), after which the events auto-expire. If there are any mistakes in our analytics-on-write processing, at best we will be able to rerun for only the last 24 hours (or one week). This is visualized in figure 11.2.

This means that we need to sit down with the bigwigs at OOPS and find out exactly what near-real-time analytics they want to see. When we do this, we learn that the priority is around operational reporting about the delivery trucks. In particular, OOPS wants a near-real-time feed that tells them the following:

This sounds like a straightforward analysis to generate. We can sketch out a simple table structure that holds all of the relevant data, as shown in table 11.1.

Now we have our five event types flowing into Kinesis, and we understand the analysis that our OOPS coworkers are looking for. In the next section, let’s define the analytics-on-write algorithm that will connect the dots.

11.1.3. Our analytics-on-write algorithm

We need to create an algorithm that will use the OOPS event stream to populate our four-column table and keep it up-to-date. Broadly, this algorithm needs to do the following:

Our approach will have similarities to the stateful event processing that we implemented in chapter 5. In this case, we are not interested in processing multiple events at a time, but we do need to make sure that we are always updating our table with a truck’s most recent location. There is a risk that if our algorithm ingests an older event after a newer event, it could incorrectly overwrite the “latest” latitude and longitude with the older one. This danger is visualized in figure 11.3.

We can prevent this by storing an additional piece of state in our table, as shown in table 11.2. The new column, Location timestamp, records the timestamp of the event from which we have taken each truck’s current latitude and longitude. Now when we process an event from Kinesis containing a truck’s location, we check this event’s timestamp against the existing Location timestamp in our table. Only if the new event’s timestamp “beats” the existing timestamp (it’s more recent) do we update our truck’s latitude and longitude.

This leaves only our last metric, Miles since oil change. This one is slightly more complicated to calculate. The trick is to realize that we should not be storing this metric in our table. Instead, we should be storing the two inputs that go into calculating this metric:

With these two metrics stored in our table, we can calculate Miles since oil change whenever we need it, like so:

For the cost of a simple calculation at serving time, we have a much easier-to-maintain table, as you will see in the next section. Table 11.3 shows the final version of our table structure, with the new Mileage and Mileage at oil change columns.

Remember that for this table to be accurate, we always want to be recording the truck’s most recent mileage, and the truck’s mileage at its last-known oil change. Again, how do we handle the situation where an older event arrives after a more recent event? This time, we have something in our favor, which is that a truck’s mileage monotonically increases over time, as per figure 11.4. Given two mileages for the same truck, the higher mileage is always the more recent one, and we can use this rule to unambiguously discard the mileages of older events.

Putting all this together, we can now fully specify our analytics-on-write algorithm in pseudocode. First of all, we know that all of our analytics relate to trucks, so we should start with a filter that excludes any events that don’t include a vehicle entity:

Now let’s process our vehicle’s mileage:

Most of this pseudocode should be self-explanatory. The trickiest part is the table[vin] .mileage syntax, which references the value of a column for a given truck in our table. The important thing to understand here is that we update a truck’s mileage in our table only if the event being processed reports a higher mileage than the one already in our table for this truck.

Let’s move on our truck’s location, which is captured only in events relating to a truck arriving or departing:

The logic here is fairly simple. Again, we have an if statement that makes updating the location conditional on the event’s timestamp being newer than the one currently found in our table. This is to ensure that we don’t accidentally overwrite our table with a stale location.

Finally, we need to handle any oil-change events:

Again, we have a “guard” in the form of an if statement, which makes sure that we update the Mileage at oil change for this truck only if this oil-change event is newer than any event previously fed into the table. Putting it all together, we can see that this guard-heavy approach is safe even in the case of out-of-order events, such as when an oil-change event is processed after a subsequent truck-departs event. This is shown in figure 11.5.

If OOPS were a real company, it would be crucial to get this algorithm reviewed and signed off before implementing the stream processing job. With analytics-on-read, if we make a mistake in a query, we just cancel it and run the revised query. With analytics-on-write, a mistake in our algorithm could mean having to start over from scratch. Happily, in our case, we can move swiftly onto the implementation!

11.2. Building our Lambda function

Enough theory and pseudocode—in this section, we will implement our analytics-on-write by using AWS Lambda, populating our delivery-truck statuses to a table in DynamoDB. Let’s get started.

11.2.1. Setting up DynamoDB

We are going to write a stream processing job that will read OOPS events, cross-check them against existing values in our table, and then update rows in that table accordingly. Figure 11.6 shows this flow.

We already have a good idea of the algorithm we are going to implement, but rather than jumping into that, let’s work backward from the table we need to populate. This table is going to live in Amazon’s DynamoDB, a highly scalable hosted key-value store that is a great fit for analytics-on-write on AWS. We can create our table in DynamoDB by using the AWS CLI tools like so:

When creating a DynamoDB table, we have to specify only a few properties up-front:

Our table has now been created. We can now move on to creating our AWS Lambda function.

11.2.2. Introduction to AWS Lambda

The central idea of AWS Lambda is that developers should be writing functions, not servers. With Lambda, we write self-contained functions to process events, and then we publish those functions to Lambda to run. We don’t worry about developing, deploying, or managing servers. Instead, Lambda takes care of scaling out our functions to meet the incoming event volumes. Of course, Lambda functions run on servers under the hood, but those servers are abstracted away from us.

At the time of writing, we can write our Lambda functions to run either on Node.js or on the JVM. In both cases, the function has a similar API, in pseudocode:

This function signature has some interesting features:

Figure 11.7 illustrates these two features of a Lambda function.

Does this function API look familiar? Remember back in chapter 5 when we wrote event-processing jobs using Apache Samza, the API looked like this:

In both cases, we are implementing a side-effecting function that is invoked for incoming events—or in Samza’s case, a single event. And as a Lambda function is run on AWS Lambda, so our Samza jobs were run on Apache YARN: we can say that Lambda and YARN are the execution environments for our functions. The similarities and differences of Samza and Lambda are explored further in table 11.4; this table shows more similarities than we might expect.

We are going to write our Lambda function in Scala, which is supported for JVM-based Lambdas alongside Java. Let’s get started.

11.2.3. Lambda setup and event modeling

We need to build our Lambda function as a single fat jar file that contains all of our dependencies and can be uploaded to the Lambda service. As before, we will use Scala Build Tool, with the sbt-assembly plugin to build our fat jar. Let’s start by creating a folder to work in, ideally inside the book’s Vagrant virtual machine:

Let’s add a file in the root of this folder called build.sbt, with the contents as shown in the following listing.

To handle the assembly of our fat jar, we need to create a project subfolder with a plugins.sbt file within it. Create it like so:

If you are running this from the book’s Vagrant virtual machine, you should have Java 8, Scala, and SBT installed already. Check that our build is set up correctly:

Now we can move on to our event-handling code. If you look back at our analytics-on-write algorithm in section 11.1.3, you’ll see that the logic is largely driven by which OOPS event type we are currently processing. So, let’s first write some code to deserialize our incoming event into an appropriate Scala case class. Create the following file:

This file should be populated with the contents of the following listing.

Let’s use the Scala console or REPL to check that this code is working correctly:

Great—we can see that a byte array representing a Truck arrives event in JSON format is being correctly deserialized into our TruckArrivesEvent case class in Scala. Now we can move onto implementing our analytics-on-write algorithm.

11.2.4. Revisiting our analytics-on-write algorithm

Section 11.1.3 laid out the algorithm for our OOPS analytics-on-write by using simple pseudocode. This pseudocode was intended to process a single event at a time, but we have since learned that our Lambda function will be given a microbatch of up to 100 events at a time. Of course, we could apply our algorithm to each event in the microbatch, as shown in figure 11.8.

Unfortunately, this approach is wasteful whenever a single microbatch contains multiple events relating to the same OOPS truck, as could easily happen. Remember how we want to update our DynamoDB table with a given truck’s most recent (highest) mileage? If our microbatch contains 10 events relating to Truck 123, the naïve implementation of our algorithm from figure 11.8 would require 10 conditional writes to the DynamoDB table just to update the Truck 123 row. Reading and writing to remote databases is an expensive operation, so it’s something that we should aim to minimize in our Lambda function, especially given that each Lambda function call has to complete within 60 seconds (configurable up to 15 minutes).

The solution is to pre-aggregate our microbatch of events inside our Lambda function. With 10 events in the microbatch relating to Truck 123, our first step should be to find the highest mileage across these 10 events. This highest mileage figure is the only one that we need to attempt to write to DynamoDB; there is no point bothering DynamoDB with the nine lower mileages at all. Figure 11.9 depicts this pre-aggregation technique.

Of course, the pre-aggregation step may not always bear fruit. For example, if all 100 events in the microbatch relate to different OOPS trucks, there is nothing to reduce. This said, the pre-aggregation is a relatively cheap in-memory process, so it’s worth always attempting it, even if we prevent only a handful of unnecessary DynamoDB operations.

What format should our pre-aggregation step output? Exactly the same as the format of our data in DynamoDB! The pre-aggregation step and the in-Dynamo aggregation differ from each other in terms of their scope (100 events versus all-history) and their storage mechanism (local in-memory versus remote database), but they represent the same analytics algorithm, and a different intermediate format would only confuse our OOPS colleagues. Table 11.5 reminds us of the format we need to populate in our Lambda function.

We should be ready now to create our first analytics-on-write code for this AWS Lambda. Create the following file:

This file should be populated with the contents of the following listing.

There’s quite a lot to unpack in our aggregator.scala file; let’s make sure you understand what it’s doing before moving on. First, we have the imports, including aliases for our event types to make the subsequent code fit on one line. Then we introduce a case class called Row; this is a lightly adapted version of the data held in table 11.5. Figure 11.10 shows the relationship between table 11.5 and our new Row case class; we will be using instances of this Row to drive our updates to DynamoDB.

Moving onto our Aggregator object, the first function we see is called map. This function transforms any incoming OOPS event into either Some(Row) or None, depending on the event type. We use a Scala pattern match on the event type to determine how to transform the event:

If you are unfamiliar with the Option boxing of the Row, you can find more on these techniques in chapter 8.

The second public function in the Aggregator is called reduce. This takes a list of Option-boxed Rows and squashes them down into a hopefully smaller list of Rows. It does this by doing the following:

Phew! If this seems complicated, it’s because it is complicated. We are rolling our own map-reduce algorithm to run against each microbatch of events in this Lambda. With the sophisticated query languages offered by technologies such as Apache Hive, Amazon Redshift, and Apache Spark, it’s easy to forget that, until relatively recently, coding MapReduce algorithms for Hadoop in Java was state-of-the-art. You could say that with our Lambda, we are getting back to basics in writing our own bespoke map-reduce code.

How do we squash a group of Rows for a given truck down to a single Row? This is handled by our reduceLeft, which takes pairs of Rows and applies our merge function to each pair iteratively until only one Row is left. The merge function takes two Rows and outputs a single Row, combining the most recent data-points from each source Row. Remember, the whole point of this pre-aggregation is to minimize the number of writes to DynamoDB. Figure 11.11 shows the end-to-end map-reduce flow.

That completes our map-reduce code for pre-aggregating our microbatch of events prior to updating DynamoDB. In the next section, let’s finally write that DynamoDB update code.

11.2.5. Conditional writes to DynamoDB

Thanks to our merge function, we know that the reduced list of Rows represents the most recent data-points for each truck from within the current microbatch. But it is possible that another microbatch with more recent events for one of these OOPS trucks has already been processed. Event transmission and collection is notoriously unreliable, and it’s possible that our current microbatch contains old events that got delayed somehow—perhaps because an OOPS truck was in a road tunnel, or because the network was down in an OOPS garage. Figure 11.12 illustrates this risk of our microbatches being processed out of order.

As a result, we cannot blindly overwrite the existing row in DynamoDB with the new row generated by our Lambda function. We could easily overwrite a more recent truck mileage or location with an older one! The solution has been mentioned briefly in previous sections of this chapter; we are going to use a feature of DynamoDB called conditional writes.

When I alluded to conditional writes earlier in this chapter, I suggested that there was some kind of read-check-write loop performed by the Lambda against DynamoDB. In fact, things are simpler than this: all we have to do in our Lambda is send each write request to DynamoDB with a condition attached to it; DynamoDB will then check the condition against the current state of the database, and apply the write only if the condition passes. Figure 11.13 shows the set of conditional writes our Lambda will attempt to perform for each row.

We are ready to take the logic from figure 11.13 and implement it in Scala. For this, we will use the AWS Java SDK, which includes a DynamoDB client. To keep our code succinct, we will also use the AWScala project, which is a more Scala-idiomatic domain-specific language (DSL) for working with DynamoDB.

Create this Scala file:

This file should be populated with the contents of the following listing.

The code for interfacing with DynamoDB is admittedly verbose, but squint at this code a little and you can see that it follows the basic shape of figure 11.13. We have a Writer module that exposes a single method, conditionalWrite, which takes in a Row as its argument. The method does not return anything; it exists only for the side effects it performs, which are conditional writes of the three elements of the Row against the same truck’s row in DynamoDB. These conditional writes are performed using the private updateIf function, which helps construct an UpdateItemRequest for the row in DynamoDB.

A few things to note about this code:

That completes our DynamoDB-specific code!

11.2.6. Finalizing our Lambda

Now we are ready to pull our logic together into the Lambda function definition itself. Create our fourth and final Scala file:

This file should be populated with the contents of the following listing.

It is this function, aowlambda.LambdaFunction.recordHandler, that will be invoked by AWS Lambda for each incoming microbatch of Kinesis events. The code should be fairly simple:

Code complete! Let’s quickly check that the code compiles fine, like so:

Now we need to assemble a fat jar with all of our Lambda function’s dependencies in it. Let’s use the sbt-assembly plugin for Scala Build Tool to do this:

Great—our build has now been completed. This completes the coding required for our Lambda function. In the next section, we will get this deployed and put it through its paces!

11.3. Running our Lambda function

Much of the appeal of a system like AWS Lambda lies in its “serverless” promise. Of course, servers are involved in running our Lambda, but they are hidden from us, with AWS taking responsibility for operating our function reliably. The price we pay for outsourcing our ops is a relatively involved setup process, which we will go through now.

11.3.1. Deploying our Lambda function

Earlier in this chapter, we set up our Kinesis stream and our DynamoDB table. We now need to deploy our assembled Lambda function and wire it into our stream and table. We will perform all of this by using the AWS CLI tools. There is a lot of ceremony to step through here, so let’s get started.

Uploading to S3

First, we need to make our local fat jar accessible to the AWS Lambda service. We do this by uploading the file to Amazon S3. This is straightforward using the AWS CLI. First, we create the bucket:

And next we upload the jar file to S3:

Unfortunately, there is a bug with AWS whereby uploading large files to newly created buckets can hang. If that happens, take a break and try again in an hour or so; you can continue with the rest of the setup in the meantime. Next, we need to configure the necessary permissions for our Lambda function.

Configuring permissions

The permissions required for operating a Lambda function are complex, so we are going to take a shortcut by using a CloudFormation template that we prepared earlier. AWS CloudFormation is a service that lets you spin up various AWS resources by using JSON templates; you can think of a CloudFormation template as the declarative JSON recipe for creating a collection of AWS services configured exactly as you want them. In Amazon parlance, we will use this template to create a stack of AWS resources—in our case, IAM roles to operate our Lambda function.

The pre-prepared CloudFormation template is publicly available in S3 in the ulp-assets bucket:

Kick off the stack creation like so:

You can monitor Amazon’s progress in creating this stack by using this command:

When you see a StackStatus of CREATE_COMPLETE in the returned JSON, we are ready to continue. We will need another piece of this output in the next section: the Output-Value for the ExecutionRole, which should start with arn:aws:iam::. Let’s create a new environment variable set to this value:

When creating this variable, make sure not to include any line breaks in the value.

Creating our Lambda function

The next step is to register our function with AWS Lambda. We can do this with a single AWS CLI command:

This command defines a function called AowLambda by using our IAM role and our fat jar previously uploaded to S3. The --handler argument tells AWS Lambda exactly which method to invoke inside our fat jar. The next three arguments configure the exact operation of the function: the function should be run against Java 8 (as opposed to Node.js), should time out after 60 seconds, and should be given 1 GB of RAM.

Attaching our function to Kinesis

Are we there yet? Not quite. We still need to identify the Kinesis stream we created earlier as the event source for our Lambda function. We do this by creating an event source mapping between the Kinesis stream and the function, again using the AWS CLI:

From the returned JSON, you can see that we have successfully connected our Lambda function to our stream of OOPS events. It reports that no records have been processed yet. In the next section, we will re-enable our event generator and check the results.

11.3.2. Testing our Lambda function

Remember that our generator script is available in the GitHub repository:

Let’s kick off our event generator:

Great! Those events are now being sent to our Kinesis stream. Let’s now take a look at our DynamoDB table. From the AWS dashboard:

You should see three rows with all of the expected fields, as shown in figure 11.14.

Leave our event generator running a little longer and then refresh the table in the DynamoDB interface. You should see the following:

Figure 11.15 shows this updated view we are expecting to see in DynamoDB.

So far, so good—we can see that our Lambda is working well to keep our operational dashboard in DynamoDB up-to-date with the latest stream of events from OOPS trucks.

The last thing we want to check is that our conditional writes are correctly guarding against old events that could arrive out of order. Remember that we don’t want to overwrite a fresh data point in DynamoDB with older data just because the older event arrives after the newer event.

To test this, first refresh the DynamoDB table to get the latest data. Then switch back to your generator’s terminal window and press Ctrl-C:

Finally, we restart the generator with a new command-line argument, backward:

As you can see, the generator is now working backward, creating ever-older events. Leave it a short while, and then head back into the DynamoDB interface and refresh the table. You’ll see that all of the data points are unchanged: the old events are not impacting on our latest truck statuses in DynamoDB at all.

For a final test, leave the backward-stepping event generator running, open a new terminal, and kick off the forward-stepping generator again:

You will have to wait a while until the event timestamps that are generated overtake your current location-timestamp values in DynamoDB, but after this happens, you should see your truck data points in DynamoDB starting to refresh again.

And that completes our testing! Hopefully, our colleagues at OOPS are pleased with the result: an analytics-on-write system that can keep a dashboard of OOPS trucks updated in near real-time from a Kinesis stream. A nice evolution for this project would be to add a visualization layer on top of the DynamoDB table, perhaps implemented in D3.js or a similar library.

Summary