Fast Data Pipeline Design: Updating Per-Event Decisions by Swapping Tables

VoltDB was one of the first companies to enable a new modern breed of applications, applications that combine streaming, or “fast data”, tightly with big data.We call these applications Fast Data Pipelines.

First, a quick high-level summary of the fast data pipeline architecture:

The first thing to notice is that there is a tight coupling of Fast and Big, although they are separate systems. They have to be, at least at scale. The database system designed to work with millions of event decisions per second is wholly different from the system designed to hold petabytes of data and generate Machine Learning (ML) models.

There are a number of critical requirements to get the most out of a fast data pipeline. These include the ability to:

• Ingest / interact with the data feed in real-time.
• Make decisions on each event in the feed in real time
• Provide visibility into fast-moving data with real-time analytics
• Seamlessly integrate into the systems designed to store Big Data
• Ability to deliver analytic results (mined “knowledge”) from the Big Data systems quickly to decision engine, closing the data loop. This mined knowledge can be used to inform per event decisions.

Hundreds of Fast Data Pipeline applications have been built and deployed using VoltDB as the fast operational database (the glue) between Fast and Big. These applications provide real-time decisioningengines in financial fraud detection, digital ad tech optimization, electric smart grid, mobile gaming and IoT industries, among others.

This blog is going to drill into how to implement a specific portion of this fast data pipeline, namely the last bullet: the ability to close the data loop, taking knowledge from a Big Data system and applying this knowledge, online, to the real-time decision engine (VoltDB).

Closing the Data Loop

“Per-event decisioning” means that an action is computed for each incoming event (each transaction).  Usually some set of facts informs the decision, often computed from historical data. These “facts” could be captured in machine learning models or consist of a set of generated rules to be executed on each incoming event. Or these facts could represented as rows in a database table, used to filter and generate optimized decisions for each event. This blog post will focus in on the latter, storing and updating facts represented in database tables.

When storing facts in database tables, each row corresponds to some bit of intelligence for a particular value or set of values.  For example, the facts might be a pricing table for airline flights, where each row corresponds to a route and service level.  Or the values might be list of demographic segmentation buckets (median income, marital status, etc) for browser cookies or device ids, used to serve up a demographic-specific ads.

Fact tables are application-specific, can be simple or sophisticated, and are often computed from an historical “big data” data set such as Spark, Hadoop, or commercial data warehouse, etc.  Fact tables can often be quite large and can be frequently recomputed, perhaps weekly, daily, or even hourly.

It is often important that the set of facts changes atomically.  In other words, if airline prices are changing for ten’s of thousands of flights, all the prices should change all at once, instantly. It is unacceptable that some transactions reference older prices and some newer prices during the period of time it takes to load millions of rows of new data.  This problem can be challenging when dealing with large fact tables as transactionally changing millions of values in can be a slow, blocking operation. Locking a table, thus blocking ongoing operations, is unacceptable when your application is processing hundreds of thousands of transactions per second.

VoltDB solves this challenge in a very simple and efficient manner.  VoltDB has the ability to transactionally swap tables in a single operation.  How this works is as follows:

1. Create an exact copy of your fact table schema, giving it a different name. Perhaps Facts_Table and Facts_Table_2.

2. Make sure the schemas are indeed identical (and neither is the source of a view).

3. While your application is running (and consulting rows in Facts_Table to make decisions), populate Facts_Table_2 with your new set of data that you wish future transactions to consult. This table can be populated as slowly (or as quickly) as you like, perhaps over the course of a day.

4. When your Facts_Table_2 is populated, and you are ready to make it “live” in your application, call the VoltDB System Procedure @SwapTables. This operation essentially switches the data for the table by swapping internal memory pointers. As such it executes in single to sub millisecond range.

5. At this point, all the data that was in Facts_Table_2 is now in Facts_Table, and the old data in Facts_Table now resides in Facts_Table_2.  You may consider truncating Facts_Table_2 in preparation for your next refresh of facts (and to reduce your memory footprint).

Let’s look at a contrived example using the VoltDB Voter sample application, a simple simulation of an ‘American Idol’ voting system. Let’s assume that each day you are going to feature different contestants for which callers can vote. Voting needs to occur 24x7, each day, with new contestants. The contestants change every day at midnight. We don’t want any downtime - no maintenance window, for example  - when changing our contestant list.

Here’s what we need to do to the Voter sample to effect this behavior:

1. First we create an exact copy of our CONTESTANTS table, calling it CONTESTANTS_2:

-- contestants_2 table holds the next day's contestants numbers -- (for voting) and names

CREATE TABLE contestants_2

(

 contestant_number integer     NOT NULL

, contestant_name   varchar(50) NOT NULL

, CONSTRAINT PK_contestants_2 PRIMARY KEY

 (

   contestant_number

 )

);

2. The schemas are identical, and this table is not the source of a materialized view.

3. The Voter application pre-loads the CONTESTANTS table at the start of benchmark with the following contestants:

1> select * from contestants;

CONTESTANT_NUMBER  CONTESTANT_NAME

------------------ ----------------

                1 Edwina Burnam   

                2 Tabatha Gehling

                3 Kelly Clauss    

                4 Jessie Alloway  

                5 Alana Bregman   

                6 Jessie Eichman  

$ cat contestants_2.csv

1, Tom Brady

2, Matt Ryan

3, Aaron Rodgers

4, Drew Brees

5, Andrew Luck

6, Kirk Cousins

$ csvloader contestants_2 -f contestants_2.csv

Read 6 rows from file and successfully inserted 6 rows (final)

Elapsed time: 0.905 seconds

$ sqlcmd

SQL Command :: localhost:21212

1> select * from contestants_2;

CONTESTANT_NUMBER  CONTESTANT_NAME

------------------ ----------------

                1 Tom Brady       

                2 Matt Ryan       

                3 Aaron Rodgers   

                4 Drew Brees      

                5 Andrew Luck     

                6 Kirk Cousins    

(Returned 6 rows in 0.01s)

4. Now that we have the new contestants (fact table) loaded and staged, when we’re ready (at midnight!) we’ll swap the two tables, making the new set of contestants immediately available for voting without interrupting the application. We’ll do this by calling the @SwapTables system procedure as follows:

$ sqlcmd

SQL Command :: localhost:21212

1> exec @SwapTables contestants_2 contestants;

modified_tuples

----------------

             12

(Returned 1 rows in 0.02s)

2> select * from contestants;

CONTESTANT_NUMBER  CONTESTANT_NAME

------------------ ----------------

                6 Kirk Cousins    

                5 Andrew Luck     

                4 Drew Brees      

                3 Aaron Rodgers   

                2 Matt Ryan       

                1 Tom Brady       

(Returned 6 rows in 0.01s)

5. Finally, we’ll truncate the CONTESTANTS_2 table, initializing it once again ready to be loaded with the next day’s contestants:

$ sqlcmd

SQL Command :: localhost:21212

1> truncate table contestants_2;

(Returned 6 rows in 0.03s)

2> select * from contestants_2;

CONTESTANT_NUMBER  CONTESTANT_NAME

------------------ ----------------

(Returned 0 rows in 0.00s)

Note that steps 3-5, loading, swapping, and truncating the new fact table, can all be done in an automated fashion, not manually as I have demonstrated with this simple example.

Running the Voter sample and arbitrarily invoking @SwapTables during the middle of the run yielded the following results:

A total of 15,294,976 votes were received during the benchmark...

- 15,142,056 Accepted

-   152,857 Rejected (Invalid Contestant)

-        63 Rejected (Maximum Vote Count Reached)

-         0 Failed (Transaction Error)

Contestant Name Votes Received

Tom Brady      4,472,147

Kirk Cousins 3,036,647

Andrew Luck      2,193,442

Matt Ryan      1,986,615

Drew Brees      1,963,903

Aaron Rodgers 1,937,391

The Winner is: Tom Brady

Apologies to those not New England-based! As you might have guessed, VoltDB’s headquarters are based just outside of Boston, Massachusetts.

Just the Facts, Ma’am

Leveraging big data intelligence to make per-event decisions is an important component of a real-time decision engine within your data pipeline. When building fast data pipeline applications using VoltDB, VoltDB provides tools and functionality to make this process easy and also painless to a running application. Two key tasks need to be performed: loading your new fact table into VoltDB, and atomically making that new data “live” to your business logic.

Loading data into VoltDB from an external data source can be done easily via a couple of approaches: you can use one of our loaders such as the CSV, Kafka or JDBC loader; or you can write an application to insert the data.

Swapping tables in VoltDB is a trivial exercise with the @SwapTable system procedure. And most importantly, swapping in new fact table data does not impact ongoing stream processing.

Streaming Data: Billions and Billions (of transactions)

 This post originally appeared on VoltDB.com in October, 2016.

Billions and billions (of transactions)

Billions and billions^[1] of transactions are executed each and every day by production-deployed VoltDB applications.

A back-of-the-napkin calculation puts the number north of 50 billion a day. The number is likely much, much higher. These billions of transactions are executed across a variety of innovative applications, applications not practical, or sometimes even possible, to build a decade ago. Chances are that you have touched VoltDB – without knowing it – today!

• When you call someone on your cell, there’s a good chance a VoltDB application is approving the call.
• If you have played some of the top 10 mobile games within the past few years, a VoltDB application likely processed your in-game events.
• If you’ve run a road race recently, there’s a good chance a VoltDB application tracked your progress.
• If you live in the UK and turn on a light switch, you are interacting with VoltDB as your electrical smart grid uses VoltDB to authorize and process grid and meter events.
• If you play baccarat in a casino, there’s a good chance the bets placed and movement of chips on each gaming table is validated by VoltDB via in-chip RFID technology.
• If you bought something in China with your credit card, it is likely that your transaction was verified as valid or detected as fraudulent by a VoltDB application.

The VoltDB engineering team has been building the world’s fastest in-memory database for over eight years now. It’s been a fun as well as challenging journey and we’ve learned a lot in the process. Some of the more notable learnings have been:

Batch is so 2000s. Legacy applications are moving from batch processing of feeds to real-time processing. New applications are being built to ingest, transact, and compute analytics on data streaming from all types of sources, such as IoT sensors, mobile devices, logs, etc. The world is moving to real-time and turning to products like VoltDB, which has the capability to handle high throughput (thousands to millions of events per second).

SQL [still] rules the roost.  VoltDB was founded around the same time as the NoSQL movement. At the time, building a distributed SQL data store seemed contrarian, given the NoSQL hype. Fast forward a half-dozen years and we now see many of the more popular NoSQL products offering SQL or SQL-like interfaces. Even the Hadoop ecosystem has added SQL interfaces to the data lake. SQL is here to stay(and, in fact, never left).

VoltDB’s commitment to SQL remains strong. Over the past year we’ve added windowing analytic SQL, geospatial SQL support (points and polygons), continuous queries via real-time materialized views, the ability to join streams of data to compute real time analytics, and the ability to very quickly generate an approximate count of the number of distinct values. More advanced SQL capabilities are made available nearly every month.

Expect the unexpected, operationally. When you build a highly-available, clustered database that runs 24×7 at high throughput, on bare metal or in the cloud, all types of error scenarios that could happen, dohappen. Disks fail, networks break down, power goes out, mistakes are made, lightning strikes.

Building a production-ready operational database is tough work. The bar is set high and not negotiable. The following features are table stakes for current and next generation applications:

• High availability: Machines, containers and virtual machines can and do break, crash or die. Networks pause or partition. The database must handle these situations properly.
• Correctness: Accuracy matters. Every client should see a consistent view of the data.
• Predictable latency: Response time should be predictable, within 99.999% of the time.
• Durability: Never lose data even in the face of nodes crashing or networks partitioning.
• Performance: The database needs to scale with your application. Adding additional nodes should scale throughput smoothly and linearly.

To maintain these requirements across all scenarios, VoltDB Engineering constantly pounds on the database in our QA lab. We focus on randomizing errors of all types. The more we learn, the more we feel we need to test more. That’s one of the reasons we reached out to Kyle Kingsbury, (aka Aphyr), to independently apply his dastardly Jepsen test suite and methodology to VoltDB. As part of this effort, we found a few issues in the product before they caused issues in customer deployments. This effort remains one of the team’s proudest accomplishments of 2016.

But we know there’s always more work to do, and we’re committed to doing it.

Streaming analytics and operational data stores are merging into a new real-time platform

The architecture required for real-time processing of streaming data is evolving rapidly. When we started building VoltDB, a developer had to cobble together numerous pieces of technology, (e.g. ZooKeeper, Kafka, Storm, Cassandra), to process real-time streams of data.

VoltDB Founding Engineer John Hugg stated at a recent Facebook @Scale conference, “OLTP at scale looks a lot like a streaming application.”  Today we’re seeing the result of that evolution: a convergence of real-time ingestion, streaming analytics, and operational interaction into a single platform. While creating the next generation of database, the VoltDB engineering team has assembled the core components for building streaming fast data applications in VoltDB:

• Real-time ingestion with in-process, highly-available importers reading from Kafka, Kinesis and other streaming sources;
• Streaming analytics on live data via ad hoc SQL analytics as well as continuous queries; and
• A blazingly fast in-memory storage and ACID transaction execution engine capable of accessing hot and historical data to make per-event decisions millions of times per second.

The VoltDB engineering team is not done building (or learning)! Our mission is to make building highly-reliable, high throughput fast data applications not only possible, as it is today, but also extremely easy. So stay tuned. Over the coming months we’ll be delivering additional geo-distributed and high availability features, additional importers and exporters for building fast data pipeline applications, and improving our real-time analytical SQL capabilities.

^[1] With a hat tip to Johnny Carson and Carl Sagan

Simplifying the Lambda Architecture

 This post originally appeared on VoltDB.com in December, 2014. 

Evolving and Simplifying the Lambda Architecture

Introduction

The Lambda Architecture defines a robust framework for ingesting streams of Fast Data while providing efficient real-time and historical analytics. In Lambda, immutable data flows in one direction: into the system. The architecture’s main goal is to execute OLAP-type processing faster - in essence,  reduce columnar analytics from every couple of seconds to 100ms or so, without actually enabling interesting new applications like real time application of user segments/scoring. As Lambda was conceived, it wasn’t designed to transact and make per-event decisions on the Fast Data, nor to be responsive to the events coming in, as they arrive.

What is Lambda?

The Lambda Architecture is a new Big Data architecture designed to ingest, process and query both fresh and historical (batch) data in a single data architecture. In his book “Big Data - Principles and best practices of scalable realtime data systems”, Nathan Marz introduces the Lambda Architecture and states that:

“The Lambda Architecture.. provides a general purpose approach to implementing an arbitrary function on an arbitrary dataset and having the function return its results with low latency.”

Nathan further defines the system as having both a batch (historical) layer, as well as a speed layer:

“The Lambda Architecture solves the problem of computing arbitrary functions

on arbitrary data in realtime by decomposing the problem into three layers: the

batch layer, the serving layer, and the speed layer. “

The batch layer is usually a “data lake” system like Hadoop, though it could also be an OLAP data warehouse like Vertica or Netezza. This historical archive is used to hold all of the data ever collected. The batch layer supports batch query; batch processing is used to generate analytics, either predefined or ad hoc.

The speed layer is defined as a combination of queuing, streaming and operational data stores. In the Lambda Architecture, the speed layer is similar to the batch layer in that it computes similar analytics - except that it computes those analytics in real-time on only the most recent data. The analytics the batch layer calculates, for example, may be based on data one hour old. It is the speed layer’s responsibility to calculate real-time analytics based on fast moving data - data that is zero to one hour old.

As you can see, if you combine the analytics produced by the batch layer as well as the speed layer, you have a complete view of the analytics across all data, fresh and historical.  The third layer of Lambda, the serving layer, is responsible for serving up results, combined from both the speed and batch layer.

To summarize, Lambda defines a Big Data architecture that allows arbitrary queries and computations on both fast moving data as well as historical data.

“Typical” Lambda Applications

The Lambda Architecture is an emerging paradigm in Big Data computing. As such, new Lambda-based applications are emerging seemingly weekly. However, one of the more common use cases of Lambda-based applications is log ingestion and accompanying analytics. “Logs” in this context could be general log collection, website clickstream logging, VPN access logs, or the popular Twitter tweet stream collection.

Log messages often are created at a high velocity. They are immutable and usually are time-tagged or time ordered. This is the "fast data" that is captured and harvested - it is this data that is ingested by both Lambda’s speed layer and batch layer, usually in parallel, by way of message queues and streaming systems (like Kafka and Storm). The ingestion of each log message does not require a response to the entity that delivered the data - it is a one-way data pipeline.

A log message’s final resting place is the data lake, where batch metrics are [re]computed.   The fast layer computes similar results for the most recent "window", staying ahead of the Hadoop/batch layer. 

Analytics at the speed and batch layer can be predefined or ad hoc. Should new analytics be desired, Lambda suggests that you can re-run the entire data set, from the data lake or from the original log files, to recompute the new metrics. For example, analytics for website click logs could be counting page hits and page popularity. For tweet streams they could be computing trending topics.

VoltDB and the Lambda Architecture

VoltDB, a clustered, in-memory, relational database.  It supports fast ingest of data, real-time ad hoc analytics and rapid export of data to downstream systems like Hadoop and OLAP offerings.  It fits squarely and solidly into the Lambda Architecture’s speed layer. Like popular streaming systems, VoltDB is horizontally scalable, highly available, and fault tolerant — all while sustaining transactional ingestion speeds of hundreds of thousands to millions of events per second. In the standard Lambda Architecture, the inclusion of VoltDB greatly simplifies the speed layer by replacing both the streaming and the operational data store portions of the speed layer. 

In the outlined Lambda Architecture, a queuing system like Kafka would feed both VoltDB and Hadoop, or VoltDB directly, which would then in turn immediately export the event to the data lake.

Future-proofing Lambda

As defined today, the Lambda Architecture is very focused on fast data collection and read-only queries on both fast and historical data. In Lambda, data is immutable - it never changes. Data comes into the system in streams and metrics, both historical and real time, and is calculated and maintained. External systems make use of the Lambda-based environment to query the computed analytics. These analytics are then used to alert, should metric thresholds be crossed, or harvested, for example in the case of Twitter trending topics.

When considering improvements to the Lambda Architecture, what if you could react, per event, to the incoming data stream? In essence, you’d have the ability to act on the incoming feed, in addition to performing real-time analytics.

Here at VoltDB we have a lot of experience building streaming applications for Fast Data, another term for the Lambda-defined “speed layer”. Most of our customers are building Fast Data applications, providing us with unique insight into the Lambda speed layer.  These systems ingest events from log files, the Internet of Things (IoT), user clickstreams, online game play, and financial applications. While some of these applications passively ingest events and provide real-time analytics and alerting on the data streams (in typical Lambda style), many of these applications have begun interacting with the stream, adding per-event decisioning and transactions in addition to real-time analytics.

Additionally, another characteristic of these systems is that the speed layer analytics can differ from the batch layer analytics. Often the data lake is used to mine intelligence via exploratory queries.  This intelligence, when identified, is then fed to the speed layer as input to the per-event decisions. Scott Jarr describes this fully interactive Lambda-like application evolution here, http://voltdb.com/blog/youd-better-do-fast-data-right/ in his Fast Data blog.

In this diagram you can see the additions to the architecture:

1.     Data arrives at a high rate and is ingested.  It is immediately exported to the Batch Layer, the Data Lake.

2.     Historical intelligence can be mined from the Data Lake and the aggregate “intelligence” can be delivered to the Speed Layer for per-event real-time decisioning (for instance, to determine which ad to display for a segmented/categorized web browser/user).

3.     Fast Data is either passively ingested, or a response can be computed by the new decisioning layer, using both real-time data as well as historical “mined” intelligence.

For a working code example of the simplified speed layer, reference the

VoltDB “fast data” application posted here: http://voltdb.github.io/app-fastdata.

Conclusion

The Lambda Architecture is a powerful Big Data analytics framework that serves queries from both fast and historical data.  However, the architecture emerged from a need to execute OLAP-type processing faster, without considering a new class of applications that require per-event decisioning: applications like real time application of user segments/scoring, fraud detection, denial of service attacks, policy and billing, etc. In its current form, Lambda is limited: immutable data flows in one direction, into the system, for analytics harvesting.

Adding VoltDB, a linearly scalable in-memory relational database, into the Lambda Architecture greatly simplifies the speed layer by reducing the number of components needed.

Lambda’s shortcoming is the inability to build responsive, event-oriented applications.

In addition to simplifying the architecture, VoltDB provides future-proof functionality to Lambda, specifically, the ability to execute transactions and per-event decisions on Fast Data as it arrives. Rather than a one-way streaming system feeding events into the speed layer, adding an ingestion engine like VoltDB provides developers with the ability to place applications in front of the event stream to capture value the moment the event arrives, rather than capturing value at some point after the event arrived on an aggregate-basis.

VoltDB improves the Lambda architecture by: 

●      Reducing the number of moving pieces, the products and components, needed.  Specifically, major components of the speed layer can be replaced by a single component, VoltDB.   Further, VoltDB can be used as a data store for the serving layer.

●      Enables the ability for the application to make per-event decisioning and transactional behavior, without re-implementing the architecture once deployed.

●      Providing the traditional relational database interaction model, with ad hoc SQL capabilities, on Fast Data.  Applications can use standard SQL providing agility to their query needs without requiring complex programming logic.

●      Providing access to standard analytics tooling, such as Tableau, MicroStrategy, and Actuate BIRT, on top of Fast Data.