Real-Time Stream Processing with Apache Spark Structured Streaming and Apache Kafka

The world is generating data at an unprecedented pace. According to IDC, the "Global Datasphere" will grow from 33 zettabytes in 2018 to 175 zettabytes by 2025 [1]. A large portion of this data is being generated in real-time, from sources like IoT sensors, server logs, user activity on websites, financial transactions, and more. To extract insights and value from this continuous influx of data, organizations are turning to stream processing technologies.

Stream processing is a paradigm that involves consuming data continuously from sources, processing it in real-time with low latency, and writing the results to sinks or exposing them for querying. Some of the key challenges in stream processing include:

• Handling high throughput and variety of data
• Processing events in a timely manner with minimal latency
• Ensuring fault-tolerance and exactly-once semantics
• Enabling complex windowing, aggregations and joins
• Integrating with machine learning to apply models in real-time

Various distributed stream processing frameworks have been developed to address these challenges. Apache Storm, released in 2011, was one of the earliest open-source distributed real-time computation systems. It employed a record-at-a-time model and ensured at-least-once semantics. Apache Spark Streaming, launched in 2013, leveraged Spark‘s batch processing engine to process streams as a series of micro-batches, improving throughput and latency over Storm.

In 2016, Spark introduced Structured Streaming, which uses a novel approach to model stream processing as a continuous series of batch computations on an unbounded table. It provides an easier-to-use and more expressive API that supports SQL queries, event-time windows, and integration with machine learning. Structured Streaming achieves end-to-end exactly-once guarantees through checkpointing and write-ahead logs.

Apache Flink and Google Cloud Dataflow (now Apache Beam) also emerged as true streaming engines that use a continuous operator model, rather than the micro-batch model used by Spark Streaming. They provide lower latency, but have a steeper learning curve and smaller ecosystem compared to Spark.

For ingesting data into a stream processing pipeline and providing a buffer between producers and consumers, Apache Kafka has become the de-facto standard. Kafka is a distributed commit log that provides durable storage and pub-sub messaging with high throughput and fault-tolerance. It decouples data producers from data consumers, allowing them to evolve and scale independently.

Integrating Spark Structured Streaming with Kafka

Structured Streaming provides built-in support for reading data from and writing results to Kafka. You can simply specify "kafka" as the data source format and provide the bootstrap servers and topic details. Here is what a typical architecture looks like:

graph LR
A[Data Sources] --> B[Kafka Producers]
B --> C{Kafka Cluster}
D[Spark Driver] --> E[Spark Executors]
E --> D
C --> D
D --> C
D --> F[Sinks]

The key components are:

• Data sources generate events that are published to Kafka topics using the Kafka Producer API. Sources can include web servers, IoT devices, databases, etc.
• A Kafka cluster consists of multiple broker nodes that store the published events in a fault-tolerant manner. Kafka partitions data based on keys, allowing for parallel consumption.
• A Spark driver node runs the main application and manages the Spark executors. It uses the Kafka Consumer API to subscribe to one or more Kafka topics and receives events.
• Spark executors are responsible for executing the streaming query and processing the events in parallel across the cluster. They can perform operations like filtering, aggregations, joins, and more.
• The processed results can be written out to various sinks like files, databases, or back to Kafka topics. Spark ensures end-to-end exactly-once semantics by check-pointing the offsets and state to HDFS or S3.

Here is an example of how to read events from a Kafka topic and compute a running count of events per type using Structured Streaming:

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, count

spark = SparkSession \
    .builder \
    .appName("EventCounter") \
    .getOrCreate()

event_schema = spark.read.json("event.json").schema

events = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .load()

parsed_events = events.select(
    from_json(col("value").cast("string"), event_schema).alias("event"))

event_counts = parsed_events \
    .groupBy("event.type") \
    .agg(count("*").alias("count")) \
    .orderBy(col("count").desc())

query = event_counts \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

The key steps are:

1. Create a SparkSession and define the event schema
2. Read from the Kafka topic using readStream and the "kafka" format
3. Parse the JSON string values into columns using from_json
4. Group by the event type and compute the count using groupBy and agg
5. Write the result to the console sink using writeStream
6. Start the streaming query and wait for termination

Advanced Concepts

Windowing and Watermarking

In many streaming applications, you need to perform aggregations over a sliding window of time, rather than over the entire duration of the stream. Structured Streaming supports several types of built-in windows:

• Tumbling window: Fixed-size, non-overlapping windows based on the event time or processing time
• Sliding window: Fixed-size, overlapping windows that slide by a specified interval
• Session window: Dynamically-sized, non-overlapping windows that group events from the same session

To handle late or out-of-order events, Structured Streaming uses a watermarking mechanism. Watermarking allows the engine to track the progress of event time and accordingly clean up old state. You can define a watermark by specifying a event time column and a threshold on how late events can be (e.g. 10 minutes).

Here is an example of computing the average price per product over a sliding window of 1 hour, with a slide interval of 15 minutes:

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, avg, window

spark = SparkSession \
    .builder \
    .appName("PurchaseAnalysis") \
    .getOrCreate()

purchase_schema = spark.read.json("purchase.json").schema

purchases = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "purchases") \
    .load()

parsed_purchases = purchases.select(
    from_json(col("value").cast("string"), purchase_schema).alias("purchase"))

windowed_avg_price = parsed_purchases \
    .withWatermark("purchase.timestamp", "1 hour") \
    .groupBy(window("purchase.timestamp", "1 hour", "15 minutes"), "purchase.product") \
    .agg(avg("purchase.price").alias("avg_price"))

query = windowed_avg_price \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

The key steps are:

1. Define the schema for incoming purchase events
2. Read events from the "purchases" Kafka topic
3. Apply a watermark on the timestamp column to handle late data
4. Group by a sliding window on the timestamp, along with the product
5. Compute the average price per product per window
6. Write the result to the console sink

Performance Tuning

To achieve optimal performance and scalability with Spark Structured Streaming and Kafka, you need to consider various parameters and configurations. Some key ones include:

• Partitioning: The number of Kafka topic partitions determines the maximum parallelism of consumers. You should have at least as many partitions as the number of Spark executors. Use a partitioning key that ensures even distribution of data.
• Batch interval: The interval at which Spark checks for new data from the source. Set it based on the latency requirements. A smaller interval can achieve lower latency but incurs more scheduling overhead. Typical values range from 100ms to several minutes.
• Executor cores: The number of tasks that can execute in parallel on each executor. For Kafka sources, you can set one core per topic partition. Increase cores for CPU intensive workloads.
• Compression: Enabling Kafka topic compression (e.g. with LZ4 or Snappy codec) can reduce the data size and improve throughput, at the cost of additional CPU cycles.
• Caching: Cache frequently accessed data frames using .cache() to avoid recomputing them in subsequent queries. This is especially useful when you have multiple output sinks.
• Checkpointing: Configure a reliable checkpoint location on a distributed file system like HDFS or S3. Checkpointing too frequently can cause overhead, while too infrequent checkpoints can lead to larger recovery times.

Real-World Case Studies

Several companies have successfully deployed Spark Structured Streaming and Kafka in production for various use cases. Here are a few notable examples:

• Netflix uses Kafka and Spark Streaming to ingest and process over 1.3 trillion events per day, including user actions, performance metrics, and diagnostic logs. They use Spark SQL to join and enrich the streaming data with dimensional data stored in Hive [2].

• Uber built a real-time analytics platform called AthenaX based on Structured Streaming and Kafka. It ingests data from Uber‘s Kafka clusters, computes windowed aggregations, and writes the results to Hive and Elasticsearch. Uber uses this platform to track business metrics, monitor infrastructure, and power real-time dashboards [3].

• Pinterest uses Kafka and Spark Streaming to analyze user engagement on their platform in real-time. They track metrics like pins, repins, clicks, and new signups, and use machine learning models to detect anomalies and trigger alerts. Pinterest reports processing over 2 million messages per second with this pipeline [4].

Conclusion

Stream processing is becoming an essential capability for data-driven organizations that want to extract real-time insights from their data. Apache Spark Structured Streaming and Apache Kafka provide a powerful and scalable foundation for building end-to-end streaming pipelines.

Structured Streaming offers a high-level API for expressing complex stream processing logic using SQL-like operations, while ensuring exactly-once fault-tolerance and seamless integration with batch jobs. Kafka acts as a distributed messaging layer that can durably buffer events at high throughput and enables flexible pub-sub architecture.

By following best practices around schema management, watermarking, performance tuning, and operational monitoring, you can deploy Spark and Kafka in production to enable a wide variety of real-time applications, from fraud detection to personalized recommendations.

As real-time analytics becomes democratized, we can expect to see more organizations adopt stream processing technologies to drive innovation and stay competitive. Emerging use cases like real-time machine learning, digital twins, and edge computing will further push the boundaries of what is possible with stream processing.

References

[1] IDC, "The Digitization of the World – From Edge to Core", Nov 2018
[2] Netflix Technology Blog, "Stream Processing with Spark Streaming and Kafka", May 2018
[3] Uber Engineering, "AthenaX: Uber Engineering‘s Open Source Streaming Analytics Platform", Apr 2019
[4] Pinterest Engineering Blog, "Real-time Analytics at Pinterest using Spark Streaming and Kafka", Dec 2018