Make your life as a data scientist easier by using this architecture to cleanse and merge your real-time and historical data in one place. Forget about accessing multiple systems and cleaning routine tasks for your analytics.
Introduction
Data-driven decisions and applications are the core of future businesses. Getting insights from your data means cost reduction, efficiency increase, and strategic advantages. More and more companies are generating and collecting streaming or real-time data next to classical batch data in databases by applying a data-centric architecture. So, companies have the challenge to cope with both streaming and batch data in their analytics to get holistic and up-to-date insights.
Streaming data and batch data must be merged before they can be visualized and processed as a combined dataset. This article describes an implementation of the Lambda Architecture based on Apache Flink and the Lambda Architecture to address exactly this challenge. Apache Flink is a framework for computations over unbounded and bounded data streams. Flink provides multiple APIs at different levels of abstraction and offers dedicated libraries for different use cases. To make things more concrete, I will use an end-to-end example throughout the article.
Starting Point
We use a data set containing event data of an air measuring station. In this set, we find hourly measurements of nine sensors for different substances in the air. Besides these, the date, time, temperature, and air humidity are recorded. The data set is published by De Vito et al. on the UCI Machine Learning Repository.
Where are we?
We assume, that the measuring station as an IoT device publishes the data every hour into an Apache Kafka topic where we can consume it. The data persists in the Apache Kafka topic for some fixed amount of time before it gets deleted. Because we don’t want to lose this historical data, we store it regularly in a persistent database to process data further at any time. These two systems build our hybrid data storage architecture.
As data scientists, we want to access all the data – real-time data from Kafka and historical data from a database – for real-time access to all data and comprehensive analysis. Normally we would have to query two systems with different access methods, which is neither convenient nor scalable nor easy to maintain.
Where do we want to be?
That’s why we need a combined view with data from Kafka AND the database or even more systems where we only need to access one system to get all data at once from history to the most recent data points. Because it is the most widely used technology, we want to use a database as our one-stop-shop for all data.
Streaming and batch data can differ in terms of schema, data types, representation of the same situation, and finally regarding their data quality. So, the data need to be standardized. Beyond, as data scientists, we also want to get rid of routine work by having the raw data automatically cleansed.
The Lambda Architecture
The system we present is based on the Lambda Architecture, first introduced by Nathan Marz in 2011. In short, the Lambda Architecture contains three relevant layers. The batch layer processes all batch data from databases or data lakes. The speed layer processes all streaming data in real-time. Thus new data is accessible with minimal latency. The serving layer is the last component and is responsible for merging the cleansed data from the other two layers and serving it to other applications. Usually, a dedicated (e.g. Java, or Python) application is used to load the data from the two layers and bring it together in a standardized form in the serving layer. With Flink, the standardization is already done in speed and batch layer. If you want to get a more detailed view of Lambda Architecture, I recommend this article.
Architecture Overview
The proposed system is based upon the Lambda architecture but solves some of its major weaknesses by using modern technologies smartly. We use mainly two tools. The first one is Apache Flink. Flink is a framework able to process streaming data AND real-time data. Therefore, it fits very well for this use case. We use Flink’s connectors to consume messages from a given Kafka topic in real-time or to read historical data via a JDBC connection from the database. There is one Flink job for the speed layer and another one for the batch layer. They differ only regarding the data ingestion. The parts for processing and output are completely identical for one use case. We will have a look at the data processing later in an example. Flink can output data to various data sinks. For our system, we use an Apache Cassandra database because it is optimized for heavy writing workloads.
Benefits
You can read a great introduction to Apache Flink here. We stated we want to standardize and prepare data.
With Flink, we can use the same code for processing batch data and streaming data, because Flink handles batch data as a finite stream of data messages.
This is a unique selling point in favor of Flink against similar options like Spark, or Storm. Furthermore, Flink gives us a lot of flexibility for complex transformation as we can use Java, Python, and even SQL to process the data. Apache Flink is built out of multiple components, which each of them contains specific functions for different data structures (data stream vs. table) or application-specific requirements (Machine Learning).
Example: Prepare the data
During an exploratory data analysis (which I skip for this article), we realize that · time and date are in the wrong format · there are many error values, which we should fix somehow · the known error value is "-200.0" · one of the attributes (NMHC(GT)) has so many error values that we need to remove the whole column
To prepare and merge the data, we follow the following path: In the next figure, you see that we have two pipelines, one for streaming data and one for batch layer. The following part of this article is divided into "Load the data", where we will use Kafka and JDBC connectors. Then "Process the data", where we use "map" and "filter" functions and even window functions to clean and standardize the data. The last step in this pipeline is of course "Merge and provide" the data. Flink consigns the processed data to the database where it is merged and stored for further use.
In this use case, all data in Kafka is defined through Apache Avro schemas. Avro is like Thrift or Protocol Buffers. Avro as a serialization system is used in many production environments because serialized data is smaller. Avro has rich data structures and increases transmission and processing efficiency.
Load Data
Let’s start by importing the data. We establish a Kafka Consumer in the Flink job for the speed layer. We need to provide the topic name, an address of a Kafka node, and some connections properties, such as the schema registry URL. I moved the consumer and producer code to a dedicated java class (KafkaConnection), so we can use it easily in more jobs.
public class KafkaConnection {
public static <avroSchema> FlinkKafkaConsumer010
getKafkaConsumer(Class avroSchemaClass, String inTopic, String
schemaRegistryUrl, Properties properties) {
FlinkKafkaConsumer010<avroSchema> kafkaConsumer = new
FlinkKafkaConsumer010<avroSchema>(
inTopic,
(DeserializationSchema<avroSchema>)
ConfluentRegistryAvroDeserializationSchema.forSpecific(
avroSchemaClass, schemaRegistryUrl),
properties);
kafkaConsumer.setStartFromLatest();
return kafkaConsumer;
}
}This Kafka consumer sends the gathered messages as defined in the Avro schema _AirQualityvalue into a data stream. This consumer builds the source for the data stream in the Flink job.
// Initialize KafkaConsumer
FlinkKafkaConsumer010 kafkaConsumer = KafkaConnection.getKafkaConsumer(AirQuality_value.class,
inTopic,
schemaRegistryUrl,
$properties);// Set KafkaConsumer as source
DataStream<AirQuality_value> inputStream = environment.addSource(kafkaConsumer);For reading out of the PostgreSQL database, we define a JDBC connector. Therefore, we use the JdbcInputFormatBuilder of Flink’s JDBC connector package and pass the necessary parameters as you can see below. Thus, we create a data stream with a Row schema.
JdbcInputFormat.JdbcInputFormatBuilder inputBuilder = JDBCConnection.getSource($fieldTypes,
$driverName,
$dbURL,
$sourceDB,
$selectQuery,
$dbUser,
$dbPassword);DataStream<Row> inputStream = environment.createInput(inputBuilder.finish());Before we can transform the data equally in both layers, we need to convert the data in the speed layer from the Avro schema to the more generic Row schema. You see this step in the figure above as the first action in the speed layer. Then data streams in both batch and speed layers have the same internal format and we can transform them the same way.
Transform Data
We transform data by applying modular transformers, each of which performs a specific operation. We set up transformers for filtering messages out of the stream, others manipulate values, change single data types or delete attributes. Additionally, there are also windowing and SQL transformations, which remove duplicates for example. We implement these transformers as java classes letting us adjust the exact behavior by providing some parameters. They are designed to be called by the filter(), map(), or window() functions of a DataStream. Of course, these transformers are basic but we can stack them and build powerful transformation pipelines.
To process the air quality data, we concatenate the "Date" and the "Time" attribute (at index 0 and 1) to one attribute via FeatureAggregation() first. Then we convert the resulting string to a SQL-ready timestamp in the second line. Next, we remove the original date and time attributes as they are not needed anymore. Moreover, we remove the NMHC(GT) attribute, which simply has too many errors and cannot be cleaned by calling the FeatureSelection transformer. The transformer only keeps the new timestamp attribute and all attributes except NMHC(GT).
DataStream<Row> timestampStream = inputstream
.map(new FeatureAggregation(0, 1, "+"))
.map(new StringToDate(14, "dd/MM/yyyy HH.mm.ss"))
.map(new FeatureSelection($keptAttributes);Now it gets slightly more advanced: We use a window function to replace error values (-200.0) with an average calculated over a time window. You might know this method in SQL as the "OVER" clause in combination with "ROWS PRECEDING". We apply this transformer to all attributes containing sensor data. Before, we need to create windows, so the messages in the data stream can be split on different windows. Like in SQL we need a key to control the window split. We define the "size" of the windows to be 20 seconds: With this configuration, all messages from the same day, which arrive within 20 secs will be routed to the same window. For each attribute, an average is calculated which replaces all error values.
DataStream<Row> windowedStream = timestampStream
.keyBy(new DateAsKey())
.window(EventTimeSessionWindows.withGap(Time.seconds(20)))
.process(new ReplaceMultipleByAverage(replace_fields,"==",
-200.0f));Unfortunately, this is not sufficient to clean the data completely. Some of the attributes contain so many errors that the calculated average in the current window is corrupted by error values, too. Therefore, we need to filter all remaining messages with errors out of the data stream.
DataStream<Row> outputStream = windowedStream.filter(
new RemoveErrorValues(0,"==", -200.0f));for (int i = 1; i < outputStream.getType().length(); i++) {
outputStream.filter(new RemoveErrorValues(i,"==", -200.0f));
}We check all fields by applying the filter function multiple times in a loop. Finally, the data is ready to be written to the database and used in further applications.
Merge and provide Data
The only thing left is to write the data out to the database. Writing it to the serving layer merges streaming and batch data at the same time: It is merged by writing it in one single table. We use the following Cassandra sink to output processed streams to the same Cassandra table. But you can also use a JDBC sink.
CassandraSink.addSink(outputStream)
.setClusterBuilder(
new ClusterBuilder() {
@Override
public Cluster buildCluster(Cluster.Builder builder) {
Cluster cluster = null;
try {
cluster = builder.addContactPoint(cassandraURL)
.withPort(cassandaPort)
.withCredentials(
cassandraUser,
cassandraPassword)
.build();
} catch (Exception e) {
e.printStackTrace();
}
return cluster;
}
}
)
.setQuery(insertQuery)
.build();The insertQuery is just a normal SQL "Insert…" query. It’s important to understand that the batch layer only writes historical data and the speed layer should have written it already some time ago when it was current. As a result, the batch layer overrides data from the speed layer because it is more accurate and trustworthy. You need to make sure the speed layer does not insert any data with existing primary keys, while the batch layer must be able to overwrite existing rows with the same primary key. This can be achieved by configuring the primary key constraint in the table and defining the action on constraint violations for each job. With this strategy, we only need one standard table and no service for merging. The data is ordered by primary keys and there aren’t any conflicts or duplicates. All Services that want to consume the preprocessed data can access the DBMS via classic methods (Connectors, REST APIs, etc.).
Conclusion
The article explained how you can process your real-time and historical data with only four tools and some reusable java code.
You do not need a sophisticated pipeline, to have a scalable and secure solution that cleans and processes your raw data and serves it to your analytics.
In short, use Kafka topics for your real-time event data, the persistent data store of your choice for historical data, Apache Flink for processing, and someplace to store output data until you analyze it. This architecture will help you to get to the value from your data much faster and operationalize your data cleaning and preparation effectively.
Disclaimer: I conducted this work as my bachelor thesis with the support of Hewlett Packard Enterprise and Dr. Bernd Bachmann.
