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.

Top 5 Ways to Better Use Your Data

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

 Top 5 Ways to Better Use Your Data

Sir Francis Bacon is said to have coined the phrase “scientia potentia est”, translated as ‘knowledge is power’. Four hundred years later, we might rewrite the phrase as “data potentia est”.  Data is power - so how can you better use your organization’s data?  Here are five suggestions to help you drive value from your organization’s data - quickly.

1)    Know what you have.

First and foremost, inventory your data. 

There are two types of data to identify. First is historical data. Historical data is data you’ve accumulated over years of doing business. This could include databases, files, spreadsheets, presentations, transactions, logs, etc. The second type of data is the data that is being created “right now” - this is real-time data. Real-time data potentially has immediate value, and then ultimately turns into historical data.

Catalog and prioritize the data you have. Ideally, you will also want to identify the sources of data. Knowing how data is created allows you to capture it, store it, and eventually extract value from it, at the least cost to the organization, and with the best ROI.

The value of each type of data is different. Historical data allows you to analyze and mine past events. Real-time data gives you the opportunity to calculate analytics, possibly compare them to historical trends, and perform business actions in real-time, to capture additional and immediate value.

By way of example, consider a fraud prevention offering. Fraudulent transaction patterns are mined from historical data. These historical patterns are applied to real-time transactions to identify and reject suspected fraudulent transactions.

2)    Architect a data strategy that handles both Big and Fast data.

Creating a historical archive, perhaps a data lake via a Hadoop cluster, to store your data is only one step. Today enterprises create data at a tremendous - and growing - rate.  Processing and ingesting data in batch mode overnight is no longer acceptable. Real-time responsive enterprises need to process and react to data in seconds to minutes. Many organizations, including mobile operators, telecom providers, financial services organizations and advertising technology providers must respond in milliseconds.

3)    Choose the appropriate technologies.

There are a plethora of big data tools, and most are designed around best practices, optimized to extract value from both historical and real-time data.

Minimally you will need technologies for these areas:

Big Data: Typically, the main data management platform for big data is Hadoop or a data warehouse or perhaps a combination of the two to handle both structured and unstructured data. They act as the repository for all your data, often called the “data lake”. The data lake stores historical data to be analyzed and mined. 

Fast Data: Data is being created at a dizzying rate every day. Fast data is data that is being created now and is streaming into your company now.  It could be user clicks on your corporate web page or product downloads or any operational event occurring in your organization.  To deliver this fast data to the systems that can act on it, consider a message queueing systems such as Kafka.  To eliminate batch processing (slow data!) this message queue needs to deliver event data to an operational data stores capable of handling and processing messages at web-scale speed, thousands to tens of thousands to even millions of events per second.  The operational data store’s role is to ingest the data and process it in real-time.  Real-time processing can include computing real-time analytics, such as counts, aggregations and leaderboards, issuing real-time alerts, deduping, enriching and aggregating events, and making transactional decisions on an event-by-event basis. Both NewSQL and NoSQL operational stores can provide horsepower for handling real-time processing of event streams.  Modern operational data stores range from strongly-consistent SQL databases to eventually consistent key/value and document stores.  Consider numerous factors when choosing, including transactions as well as query interface, important for your data visualization tooling.

Data Visualization: Dashboards, charts, leaderboards, pivot tables, and visualizations all play a key role in understanding your data, both historical and real-time.

Historical visualization helps you explore, understand patterns, and create predictive analytics. Real-time visualizations help you understand the current state of your business, usually in the form of a real-time dashboard.

You will want to evaluate tools from vendors such as Tableau, Qlik and MicroStrategy for dashboarding and ad hoc visualizations -- user experience is a critical factor with this kind of software so having your users try it out is essential.

Data Science: A growing number of tools can help you extract information and insight from your data. Machine learning packages provide data classification, clustering, and regression analysis, and allow software to “learn” to identify and make predictions on data. Consider popular open source offerings such as Spark (MLlib) or R to get started.

4)    Build a Data Pipeline that delivers Data as a Service (DaaS) to internal customers.

Define an architecture that serves data to your internal customers. Capturing and analyzing the data is great, but it is only the first step. Data and insights must be readily available to consumers (people and applications) across your enterprise. Consumers of your data must be able to tap into both historical data from the data lake as well as real-time fast data, along with the insights derived from both together.

5)    Begin building applications to extract value from the data  -  then iterate.

Start small and add incrementally. Identify opportunities for small quick wins that will prove you can capture value from your data. Realize that data evolves and new patterns will emerge. Foster an environment of experimentation, innovation and continuous improvement and iterate on your data analysis.

Data is valuable. Batch processing is so 1990s. Now you’ve got five ideas for how to extract more value from your data. Start now and iterate. Think Big, of course, but also Think Fast.

3 Fast Data Application Patterns

 This post originally appeared on VoltDB.com in January, 2015.  It has been lightly updated in this post.

Three Fast Data Application Patterns

The focus of many developers and architects in the past few years has been on Big Data, specifically mining historical intelligence from the Data Lake (usually a Hadoop stack containing terabytes to petabytes of data). Now, product architects are asking how they can use this business intelligence for competitive advantage. As a result, application developers have come to see the value of using and acting in real-time on streams of fast data; using OLAP reporting wisdom, they can realize the benefits of both fast data and Big Data. As a result, a new set of application patterns have emerged. The applications are designed to capture value from fast-moving streaming data, before it reaches Hadoop.

At VoltDB we call this new breed of applications “fast data” applications. The goal of these fast data applications is to do more than just push data into Hadoop asap, but also to capture real-time value from the data the moment the data arrives. 

Because traditional databases historically haven’t been fast enough, developers have been forced to go to great effort to build fast data applications - they build complex multi-tier systems often involving a handful of tools typically utilizing a dozen or more servers.  However, a new class of database technology, especially NewSQL offerings, has changed this equation.

If you have a relational database that is fast enough, highly available, and able to scale horizontally, the ability to build fast data applications becomes less esoteric and much more manageable. Three new real-time application patterns have emerged as the necessary dataflows to implement real-time applications. These patterns, enabled by new, fast database technology, are:

1.     Real-time Analytics

2.     Real-time Decision Engine

3.     Fast Data Pipeline

Let’s take a look at the characteristics of each of these fast data application patterns and how a NewSQL database like VoltDB can improve and simplify building applications.

Real-time Analytics

This application pattern processes streaming data from one or many sources and performs real-time analytic computations on that fast data. Today this application pattern is often combined with Big Data analytics, producing analytics on both fast data and big data.

In this pattern, the value captured from the fast data stream is primarily real-time analytics. The streaming engine tracks counts and metrics derived from each message. Applications tap these stored results and display dashboard state and possibly offer real-time alerts.

Important features in this application pattern include:

• Pre-built connectors are required to easily feed the streams of data into the database.
• The database needs to compute real-time analytics by pre-computing materialized views on a per-message basis.
• Standard reporting tools such as Tableau or MicroStrategy can use SQL to query real-time state and compute ad hoc analytics.
• The database needs the ability to perform analytics and aggregations over time windows of data, such as by the second, minute, hour, day, etc.
• The database needs the ability to discard, or age-out, data after it is processed.  VoltDB accomplishes this with “capped tables,” the ability to define a table constraint to limit the number of rows a table has, and when that constraint is violated, to automatically execute delete statements to remove older rows.

Real-time Decision Engine

This application pattern processes inbound requests from many clients, perhaps tens of thousands simultaneously, and returns a low latency response or decision to the client. This is a classic OLTP application pattern but running at scale against high velocity incoming data.

Scaling to support per-event high velocity transactions enables applications that evaluate campaign, policy, authorization and other business logic, to respond in real-time, in milliseconds, to applications. In this pattern, the business value is providing “smart,” or calculated, responses to high velocity requests. Applications that make use of this model today include digital ad-tech campaign balance processing as well as ad choice (based on precomputed user segmentation or other mined heuristics), smart grid electrical grids, and telecom billing and policy decisioning. In all cases, the database is processing incoming requests at exceptionally high rates. Each incoming request runs a transaction to calculate a decision and return a response to the calling application.  For example, an incoming telecom request (a new Call Data Record), may need to decide, “Does this user have enough balance to process this call?” A digital ad platform may ask, “Which of my ads should I serve to this mobile device, based on campaign available balance?”

Important features in this application pattern include:

• ACID Transactions. In this pattern, the decision engine (the database) is updating state in a consistent and durable manner. The state is often a balance of some kind (usually monetary). Additionally, consistency is important. These responses (decisions) are based on data that must be correct, thus consistent. 
• Low predictable latency responses. These applications need a response in real-time. Often there is a budget for database processing that ranges in the single-digit millisecond range, 99.999% of the time.
• Durable data. In this pattern the database is often a system of record, at least for a time window. Should the system go down, data would need to be durable and recoverable.
• Ability to use standard tooling, i.e. SQL, to query and compute real-time state.  Real-time dashboards capturing the state of the system (balances, transactions/second) as well the ability to ad hoc query state are important to this application pattern.

For a real-time decisioning (high velocity transactions) code example demonstrating an ‘American Idol’-like voting system, with per-vote validation, see https://github.com/VoltDB/voltdb/tree/master/examples/voter.

Fast Data Data Pipeline

This application pattern processes streaming data from one or many sources and performs real-time ETL (Extract, Transform, Load) on the data, delivering the result to a historical archive. In a streaming data pipeline, incoming data may be sessionized, enriched, validated, de-duped, aggregated, counted, discarded, cleansed, etc. by the database before being delivered to the Data Lake.

Important features in this application pattern include:

• Pre-built connectors to feed the streams of data into the pipeline. VoltDB includes connectors to import Kafka streams, relational data, and also supports HadoopOutputFormat results from Hive and Pig.
• The ability to process data, to aggregate, de-dupe, or transform, as part of ETL workflow. VoltDB transactions, implemented as Java Stored Procedures, allow developers to transactionally execute SQL, combined with Java business logic, to process each message individually or in aggregate.
• Pre-built export connectors to stream data downstream to historical archive as fast as it arrived. VoltDB includes export connectors to stream data to Kafka, RabbitMQ, Hadoop, Vertica, Netezza, or any relational data store via JDBC.

Applications making use of this pattern often are processing continuous streams of data that must be validated, transformed and archived in some manner. One example is processing device ids (usually in the form of cookies). The pipeline computes segmentation output intelligence, providing correlation data to be used for advanced decisioning applications, often in the digital ad tech arena. 

For a fast data pipeline code example demonstrating click stream processing based on user segmentation, using VoltDB as the ingestion and processing engine, see https://github.com/VoltDB/app-fastdata.

VoltDB and the Fast Data Pipeline

The fast data processing layer must have the following properties across all use cases:

• High ingestion (write) rate. The pipeline must ingest data at historically challenging transaction rates. Transactions occurring at hundreds of thousands to millions of times per second are not uncommon.
• Analyze incoming data in real-time. Real-time analytics enable users to derive seasonal patterns, statistical summaries, scoring models, recommendation models, rankings, leader boards and other artifacts for use in user-facing applications.
• Real-time decisions. Enabling real-time transactions against new incoming data makes it possible to respond back to users or drive downstream processes based on the output of the analysis activity and the context and content of the present event. 

In addition, the three patterns require a system architected to deliver:

• High availability. The pipeline processing engine, in this case VoltDB, can survive machine loss or (most) networking failures, either for routine maintenance or due to environmental issues or errors, and continue to operate correctly.
• Elastically and horizontally scalable. As throughput increases, more machines can be added to the pipeline processing engine to accommodate the additional traffic without interrupting the running system.

VoltDB is an in-memory, relational database that is fully durable and maintains strict ACID properties. VoltDB is highly available and elastically scales-out on commodity hardware. VoltDB includes a set of pre-built integrations, both streaming importers and exporters, all designed to help you ingest streaming data, process it within VoltDB, and, once processed, export data seamlessly to a historical data warehouse.

Whether you view your problem as OLTP with real-time analytics or as stream processing, VoltDB is the only system that combines ingestion, speed, robustness and strong consistency to make developing and supporting apps easier than ever. Everything we build enables fast data apps. We’ve helped our customers develop and deploy hundreds of them.