Moving real-time analytics to the next level

We have been running Real Time Analytics, based on Apache Kafka, Spark Streaming and Apache Druid for more than one year. After solving the problem of connecting these technologies, described in my recent blog post, we added many use-cases to our system. In general, each use-case was built according to the principle of: read raw events from one Kafka topic, process them in a Spark Streaming job, write into another Kafka topic, and ingest into Apache Druid.

It has enabled us to obtain a wide range of insights from our Real Time dashboards. However, each type of events had its own flow and had its own visualizations, independent from other types of events. For example, we had a separate view for all our clicks and for our impressions. We could create a dashboard which shows two tiles — one for clicks and one for impressions, side by side.

However, we could not get metrics based on these two streams simultaneously. For example, we could not calculate the CTR — Click_Through-Rate — which is a very important metric in the ad-tech industry. Calculating the CTR requires the number of clicks for a specific time slot divided by the number of impressions. For a relational database this is straight-forward task — run a single query selecting from two or more tables using the JOIN syntax. This is not the case in Druid. The Join operation has recently been added to the latest Druid versions, however, it is mainly used as part of the Lookups feature. Complete support of JOINs is expected to be delivered in the next versions, only towards the end of 2020.

Union vs Join

In order to maximize benefit from our Real Time analytics system, we wanted to find a way to calculate metrics based on different types of events, without using JOINs. We were looking for an out-of-the box idea. We decided to use logical UNION to tackle this problem.

Let’s consider the following example: we have a single large table that contains two types of events — clicks and impressions.

We see that for a specific time period we got three clicks and five impressions. So CTR for that period is 3/5=0.6

Similarly, we can calculate another important metric — RPM — Revenue per Mille, or the revenue per thousand impressions.

Both these metrics are calculated from a single table, without necessitating a join to other tables. The trick here is to “mark” different types of events in order to distinguish between them during the query.

realtime_event — the Hub of Real Time Data.

Based on this example, we’ve built a system that streams several different event types into a single Kafka topic, and then to Druid. This system contains several Spark Streaming jobs, each job handling a specific type of event. Each job reads event level data from the original Kafka topic, enriches it with more fields, and writes to the output Kafka topic. Note that all jobs write into the same Kafka topic. Each job adds an “event_type” field to each event, in addition to other fields, according to specific business logic. Usually jobs add some boolean fields, for example, the job that handles clicks can validate click status and set “is_valid_click” to either 0 or 1. This enables these fields to be handled efficiently in Druid and SQL queries.

Sampling the data

Processing this volume of data in Real Time is resource intensive. We need resources in our Spark Streaming cluster to run Spark Streaming jobs. We need resources in our Kafka cluster, to contain the unified topic for all types of events. Finally, we need resources in our Druid cluster to store all these events.

Let’s recap — what benefit do we want from the system? Our goal is to build a system that presents a Real Time view for our data. Our goal was to close the gap created by our batch-based main data processing system which utilizes Hive, Vertica, Tableau and hundreds of ETL jobs every hour. The batch process provides accessibility to the data with several hours latency. By contrast, the system built over Druid should be fast and flexible, and we can also allow for some data discrepancies. Therefore, we decided to reduce the load on our systems by sampling the data.

Sampling in Spark job.

We use RDD.sample API to sample the events read from the Kafka topic. Sample rate is a parameter for all our Spark jobs.

rdd.sample(false, sampleRate, seed);

The sample rate can be different per job. For example, the number of impressions is way larger than the number of clicks, so we can sample 100% of the clicks, but only 1% percent of impressions.

This technique reduces the load on Spark, Kafka and Druid. However we need to keep in mind that we have only a portion of all events in Druid. How can we guarantee that data that we query from Druid is correct and gets us the complete picture?

Dealing with sampling in Druid.

Let’s assume we sample 1% of impression events in the Spark Streaming job. Consequently in Druid we have only 1% of impressions. Note that all records for impressions have the following marker field: is_impression = 1. Usually, without sampling, we would run the following query in Druid for getting the number of impressions:

SELECT SUM(is_impression) from realtime_event;

Now, running this query on the sampled data we need to remember that the result should be multiplied by 100. What happens if we increase our sample rate from 1% to 10%? In this case, we need to multiply the query result from Druid by 10. This manual approach can work for running a few ad-hoc queries. However, it is not scalable for a system that contains multiple Spark jobs (each one with its own sample rate). Furthermore, we developed our system for multiple users from different teams — the sampling approach should be transparent to them. They should be able to query the data simply and transparently.

Adding sampleFactor

In order to enable transparent handling of the data we have in Druid, we came up with the following idea: We added a “sampleFactor” field to each event, calculated by the following formula:

sampleFactor = 1/sampleRate.

So, for a sample rate of 1% the sampleFactor is 100. Consequently measures in Druid should be multiplied by 100.

We use the sampleFactor for calculating important measures in Druid during ingestion from Kafka.

Selective sampling

Rolling out this Real Time data to production, we faced a new requirement — scenarios where specific traffic should be “fully sampled”. For example, the traffic for specific publishers. This new requirement was easily implemented using Spark streaming jobs. Firstly, we detect if the event belongs to the specific publisher, which should be fully sampled. If affirmative, we don’t apply the rdd.sample() for this event; we also add sampleFactor=1 to this event. Using this approach we still get the correct number in Druid, even when some of the events are fully sampled.

Sampling vs Rollup

Why did we develop our ad-hoc “sampling” mechanism instead of using the Druid built-in rollup feature? The reason is that the realtime_event datasource is a union of several event types. It also contains the union of various dimensions. There are more than 100 dimensions within that datasource. We can add more dimensions according to requirements with relative ease. Using Druid’s rollup feature for the number of dimensions available would not be efficient. Druid adds internal counters and sums for rollup data sources. Defining rollup for about 100 dimensions reduces the probability of two events having exactly the same dimensions (for one minute). However, the overhead of internal counters and sums still remains, making it inefficient.

Summary

By using the UNION of different events types into a single data source in Druid, we’ve built a very broad view of the data, in Real Time. This facilitates varying data analytic use cases by different teams. It also allows for aggregation of various measures that were traditionally calculated by using JOINs between tables. We handle scaling and resource problems by sampling of our events. Our sampling technique can be easily extended according to requirements. In addition, we manage the data in Druid so that the underlying sampling remains transparent to the teams consuming the Druid data.

One Event to Rule them All was originally published in TDS Archive on Medium, where people are continuing the conversation by highlighting and responding to this story.

As a Data Engineer I’m dealing with Big Data technologies, such as Spark Streaming, Kafka and Apache Druid. All of them have their own tutorials and RTFM pages. However, when combining these technologies together at high scale you can find yourself searching for the solution that covers more complicated production use-cases.
In this blogpost I’m going to share the knowledge I gained by combining Spark Streaming, Kafka and Apache Druid all together for building real time analytics dashboard, guaranteeing precise data representation.

Before we dive in…few words about Real Time Analytics

The Real-time analytics is a new trend in Big Data technologies, and usually has significant business effect. When analyzing fresh data, the insights are more precise. For example, providing real time analytics dashboard for Data Analysts, BI and Account Managers teams can help these teams to make fast decisions.
The commonly used architecture for real time analytics at scale is based on Spark Streaming and Kafka. Both these technologies are very well scalable. They run on clusters and divide the load between many machines. The output of Spark jobs can go to many various destinations, it depends on the specific use case and the architecture. Our goal was to provide the visual tool displaying real-time events. For this purpose we chose Apache Druid database.

Data visualization in Apache Druid

Druid is a high performance real-time analytics database. One of its benefits is the ability to consume real time data from Kafka topic and build powerful visualizations on top of it using Pivot module. Its visualizations enable running various ad-hoc “slice and dice” queries and get visual results quickly. It is very useful for analyzing various use cases, for example how specific campaigns perform in certain countries. Data is retrieved at real-time, with 1–2 minutes delay.

The architecture

So we decided to build our Real Time analytics system based on Kafka events and Apache Druid. We already had events in Kafka topic. However we could not just ingest them into Druid as is. We needed to add more dimensions to each event. We needed to enrich every event with more data in order to see it in Druid in a convenient way. Regarding the scale, we’re dealing with hundreds of thousands of events per minute, so we needed to use technology that can support these numbers. We decided to use Spark Streaming job for the enrichment of original Kafka events.

Figure 1. Real time analytics architecture

Spark Streaming job runs forever? Not really.

The idea of Spark Streaming job is that it is always running. The job should never stop. It constantly reads events from Kafka topic, processes them and writes the output into another Kafka topic. However, this is an optimistic view. In real life things are more complicated. There are driver failures in the Spark cluster, in which case the job is restarted. Sometimes the new version of spark application is deployed into production. What happens in this case? How does the restarted job read Kafka topic and process the events? Before we dig into these details, this figure shows what we see in Druid when the Spark Streaming job is restarted:

Figure 2. Data loss on job restart

It is definitely data loss!

What problem are we trying to solve?

We are dealing with Spark Streaming application which reads events from one Kafka topic and writes them into another Kafka topic. These events are visualized later in Druid. Our goal is to enable smooth data visualization during the restart of our Spark Streaming application. In other words, we need to ensure that no events are lost or duplicated during the Spark Streaming job restart.

It’s all about offsets

In order to understand why data is lost on jobs restart, we need to get familiar with some terms in Kafka architecture. Here you can find an official Kafka documentation. In a nutshell: events in Kafka are stored in topics; each topic has divided into partitions. Each record in a partition has an offset — a sequential number which defines the order of the record. When application consumes the topic, it can handle offsets in several ways. The default behavior is always to read from the latest offsets. Another option is to commit offsets, i.e. to persist offsets so the job can read the committed offsets on restart and continue from there. Let’s see our steps towards the solution, and get a deeper understanding of Kafka offsets management with each step.

Step#1 — auto commit offsets

The default behavior is always to read from the latest offsets. This will not work because when the job is restarted, there are new events in the topic. If the job reads from latest, it loses all messages that were appended during the restart, as can be seen in Figure 2. There is a “enable.auto.commit” parameter in Spark Streaming. Its value is false by default. Figure 3 shows the behavior after changing its value to true, running the Spark application and restarting it.

Figure 3. Data spike of job restart

We can see that using Kafka auto-commit feature causes a new effect. There is no “data loss” , however we now see duplicate events. There was no real “burst” of events. What actually happened is that auto commit mechanism commits offsets “from time to time” . There are many messages in the output topic that were not committed. After the restart the job consumes messages from the latest committed offsets and processes some of these events again. That’s why on the output we get a burst of events.

Clearly, incorporating these duplications into our visualization may mislead the business consumers of this data and have an impact on their decisions and trust in the system.

Step#2: Commit Kafka offsets manually

So we can’t rely on Kafka auto-commit feature. We need to commit Kafka offsets by ourselves. In order to do this, let’s see how Spark Streaming consumes data from Kafka topics. Spark Streaming uses an architecture called Discretized Streams, or DStream. DStream is represented by a continuous series of RDDs ( Resilient Distributed Datasets), which is one of the Spark’s main abstractions. Most Spark Streaming jobs look something like this:

dstream.foreachRDD { rdd => 
    rdd.foreach { record => process(record)} 
}

In our case processing the record means writing the record to the output Kafka topic. So, in order to commit Kafka offsets we need to do the following:

dstream.foreachRDD { rdd =>
 val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges    
 rdd.foreach { record => process(record)} 
 stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) 
}

This is a straight-forward approach and before we discuss it deeper let’s take a look at the big picture. Let’s assume that we handle offsets correctly. Namely, that offsets are saved after each processing of RDD. What happens when we stop the job? The job is stopped in the middle of the processing of RDD. The part of micro-batch is written to the output Kafka topic and is not committed. Once the job runs again, it will process some messages for the second time and the spike of duplicate messages will appear (as before) in Druid:

Figure 4. Data spike on job restart

Graceful Shutdown

Turns out there is a way to ensure a job is not killed during an RDD processing. It is called a “Graceful Shutdown”. There are several blog posts describing how you can kill your Spark application gracefully, but most of them relate to old versions of Spark and have many limitations. We were looking for a “safe” solution that works for any scale and does not depend on a specific Spark version or operating system. To enable Graceful Shutdown, the Spark context should be created with the following parameters:
spark.streaming.stopGracefullyOnShutdown = true.
This instructs Spark to shut down StreamingContext gracefully on JVM shutdown rather than immediately.
In addition, we need a mechanism to stop our jobs intentionally, for example when deploying a new version. We’ve implemented the first version of this mechanism by simply checking the existence of an HDFS file that instructs the job to shut down. When the file appears in HDFS, the streaming context will stop with the following parameters:
ssc.stop(stopSparkContext = true, stopGracefully = true)

In this case the Spark application stops gracefully only after all received data processing is completed. This is exactly what we need.

Step #3: Kafka commitAsync

Let’s recap on what we have so far. We intentionally commit Kafka offsets in each RDD processing (using Kafka commitAsync API) and we use spark graceful shutdown. Apparently, there was another caveat. Digging into the documentation of Kafka API and Kafka commitAsync() source code, I’ve learned that commitAsync() only puts the offsetRanges into a queue, which is actually processed only in the next loop of foreachRDD. Even if the Spark job is stopped gracefully and finishes processing of all its RDDs, the offsets of last RDD are actually not committed. To solve this problem, we’ve implemented a code that persists Kafka offsets synchronously and does not rely on Kafka commitAsync(). For each RDD we then stored the committed offsets in an HDFS file. When the job starts running again, it loads the offsets file from the HDFS and consumes Kafka topic from these offsets and on.

Here, it works!

It was only the combination of a graceful shutdown and a synchronous storage of Kafka offsets that provided us with the desired result. No data loss, no data spikes during restarts:

Figure 5. No data loss of spikes during Spark job restart

Conclusion

Solving the integration problem between Spark Streaming and Kafka was an important milestone for building our real-time analytics dashboard. We’ve found the solution that ensures stable dataflow without loss of events or duplicates during the Spark Streaming job restarts. We now have the trustworthy data which is visualized in Druid. Thanks to this, we’ve added more types of events (Kafka topics) into Druid and built real time dashboards. These dashboards provide insights for various teams, such as BI, Product and Customer Support. Our next goal is to utilize more features of Druid, like new analytical functions and alerts.

Originally published at https://www.outbrain.com on August 11, 2019.

Understanding Spark Streaming with Kafka and Druid | was originally published in Outbrain Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.

Working in an infrastructure team is different from working in a development team. The team develops tools that are used by developers’ teams inside Outbrain. It is very important to provide a convenient way for working with new technologies. Having a Spark cluster enables running distributed data processing in scale. Until now developers worked with Spark writing ad-hoc crontab jobs in production. It created a messy and unstable environment and required manual maintenance in case of failure. For that reason, our team wanted to give an infrastructural way for working with Spark.

In this blog post, I explain how we built the solution implementing Spark to be a native tool in our WorkflowEngine.

Meet WorkflowEngine

The Outbrain Data Infrastructure team enables running hundreds of ETL jobs every hour. These jobs are orchestrated by WorkflowEngine — an internal tool, which is developed by the team.

The WorkflowEngine (WFE) enables running various types of transformations from many kinds of data sources: MySQL, Cassandra, Hive, Vertica, Hadoop etc., located in two data centers and the cloud. Every transformation, or flow, implements some business logic. A flow may contain several tasks, which are executed one after another.

Flows are developed and owned by researchers, data scientists and developers in other groups at Outbrain, such as Algorithm research groups, BI and Data Science.

Defining flows in WFE gives users a lot of benefits. It enables users to autonomously specify dependencies between flows, so when one flow ends, another is triggered. In addition, users can define a number of retries for their flows, so if a flow fails for some reason, WFE reruns it. If all retries fail, WFE sends an email or alert about the flow failures. Logs for each running flow are shown in a convenient form in WFE UI. Furthermore, WFE collects metrics for all flows, which enables us to track the health of the system.

From the operational perspective, several instances of WFE are deployed on machines which are owned by the infrastructure team, users just define ETL flows in a separate repository.

Diving Into the Solution

In order to respond to requirements for running data analytics on Spark, our team had to allow an infrastructural way for running Spark jobs orchestrated by WFE. To achieve this, we’ve developed a new kind of WorkflowEngine task called SparkStep.

When approaching the design for SparkStep, we had several principals in mind:

1. Avoid producing additional load on WFE machines
2. Create an isolated environment for every Spark job
3. Allow users to specify and at the same time allow the system to limit the number of YARN resources allocated for every SPARK job
4. Enable using multiple Spark clusters with different versions

Taking all this into consideration, we came up with the following architecture:

Keeping the control while having fun

The Spark job definition specifies how a job will run on the Spark cluster (e.g. jarURL, className, configuration parameters). In addition, there are properties that are set internally by WFE. When WFE starts executing a SPARK job, it builds a spark-submit command using all these properties.

The spark-submit command submits the Spark job to YARN, which allocates needed resources and launches the application. After submitting the job, WFE constantly checks its status using YARN REST API. When the job is completed in YARN, WFE reports a COMPLETED status for the WFE Spark task and continues running the next task. However, if the Spark job fails in YARN, WFE reports a FAILED status for the WFE job. In this case, WFE can rerun the job, according to the defined number of retries. If the job runs in YARN for too long, and its running time exceeds the defined maximum, WFE stops the job using the YARN REST API. In this case, the job is marked as FAILED in WFE. This functionality ensures both that we have a cap on the amount of resources each flow can use and that resources are saved when a job is canceled for any reason.

The spark-submit command runs in a Docker container. The Docker exits once the job is submitted to the cluster. It enables us to create an isolated environment for every Spark job executed by WFE. In addition, using Docker container, we can easily use different Spark versions simultaneously, by creating different Docker images.

By using this architecture we achieve all our goals. Spark jobs are submitted to YARN in a “cluster” mode, so they do not create a load on WFE machines. YARN handles all resources for each job. Every Spark job runs in an isolated environment owing to the Docker container.

Originally published at www.outbrain.com on June 12, 2018.

Making Spark Native for Data Processing Framework was originally published in Outbrain Engineering on Medium, where people are continuing the conversation by highlighting and responding to this story.