A brief demonstration of Apache Spark Structured Streaming, Apache Kafka Streams, Apache Flink, and Apache Pinot with Apache Superset

Introduction

According to TechTarget, “Stream processing is a data management technique that involves ingesting a continuous data stream to quickly analyze, filter, transform or enhance the data in real-time. Once processed, the data is passed off to an application, data store, or another stream processing engine.” Confluent, a fully-managed Apache Kafka market leader, defines stream processing as “a software paradigm that ingests, processes, and manages continuous streams of data while they’re still in motion.”

Batch vs. Stream Processing

Again, according to Confluent, “Batch processing is when the processing and analysis happens on a set of data that have already been stored over a period of time.” A batch processing example might include daily retail sales data, which is aggregated and tabulated nightly after the stores close. Conversely, “streaming data processing happens as the data flows through a system. This results in analysis and reporting of events as it happens.” To use a similar example, instead of nightly batch processing, the streams of sales data are processed, aggregated, and analyzed continuously throughout the day — sales volume, buying trends, inventory levels, and marketing program performance are tracked in real time.

Bounded vs. Unbounded Data

According to Packt Publishing’s book, Learning Apache Apex, “bounded data is finite; it has a beginning and an end. Unbounded data is an ever-growing, essentially infinite data set.” Batch processing is typically performed on bounded data, whereas stream processing is most often performed on unbounded data.

Stream Processing Technologies

There are many technologies available to perform stream processing. These include proprietary custom software, commercial off-the-shelf (COTS) software, fully-managed service offerings from Software as a Service (or SaaS) providers, Cloud Solution Providers (CSP), Commercial Open Source Software (COSS) companies, and popular open-source projects from the Apache Software Foundation and Linux Foundation.

The following two-part post and forthcoming video will explore four popular open-source software (OSS) stream processing projects, including Apache Spark Structured Streaming, Apache Kafka Streams, Apache Flink, and Apache Pinot. Each of these projects has some equivalent SaaS, CSP, and COSS offerings.

This post uses the open-source projects, making it easier to follow along with the demonstration and keeping costs to a minimum. However, you could easily substitute the open-source projects for your preferred SaaS, CSP, or COSS service offerings.

Apache Spark Structured Streaming

According to the Apache Spark documentation, “Structured Streaming is a scalable and fault-tolerant stream processing engine built on the Spark SQL engine. You can express your streaming computation the same way you would express a batch computation on static data.” Further, “Structured Streaming queries are processed using a micro-batch processing engine, which processes data streams as a series of small batch jobs thereby achieving end-to-end latencies as low as 100 milliseconds and exactly-once fault-tolerance guarantees.” In the post, we will examine both batch and stream processing using a series of Apache Spark Structured Streaming jobs written in PySpark.

Apache Kafka Streams

According to the Apache Kafka documentation, “Kafka Streams [aka KStreams] is a client library for building applications and microservices, where the input and output data are stored in Kafka clusters. It combines the simplicity of writing and deploying standard Java and Scala applications on the client side with the benefits of Kafka’s server-side cluster technology.” In the post, we will examine a KStreams application written in Java that performs stream processing and incremental aggregation.

Apache Flink

According to the Apache Flink documentation, “Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.” Further, “Apache Flink excels at processing unbounded and bounded data sets. Precise control of time and state enables Flink’s runtime to run any kind of application on unbounded streams. Bounded streams are internally processed by algorithms and data structures that are specifically designed for fixed-sized data sets, yielding excellent performance.” In the post, we will examine a Flink application written in Java, which performs stream processing, incremental aggregation, and multi-stream joins.

Apache Pinot

According to Apache Pinot’s documentation, “Pinot is a real-time distributed OLAP datastore, purpose-built to provide ultra-low-latency analytics, even at extremely high throughput. It can ingest directly from streaming data sources — such as Apache Kafka and Amazon Kinesis — and make the events available for querying instantly. It can also ingest from batch data sources such as Hadoop HDFS, Amazon S3, Azure ADLS, and Google Cloud Storage.” In the post, we will query the unbounded data streams from Apache Kafka, generated by Apache Flink, using SQL.

Streaming Data Source

We must first find a good unbounded data source to explore or demonstrate these streaming technologies. Ideally, the streaming data source should be complex enough to allow multiple types of analyses and visualize different aspects with Business Intelligence (BI) and dashboarding tools. Additionally, the streaming data source should possess a degree of consistency and predictability while displaying a reasonable level of variability and randomness.

To this end, we will use the open-source Streaming Synthetic Sales Data Generator project, which I have developed and made available on GitHub. This project’s highly-configurable, Python-based, synthetic data generator generates an unbounded stream of product listings, sales transactions, and inventory restocking activities to a series of Apache Kafka topics.

Source Code

All the source code demonstrated in this post is open source and available on GitHub. There are three separate GitHub projects:

	# streaming data generator, Apache Spark and Apache Pinot examples
	git clone –depth 1 -b main \
	https://github.com/garystafford/streaming-sales-generator.git
	# Apache Flink examples
	git clone –depth 1 -b main \
	https://github.com/garystafford/flink-kafka-demo.git
	# Kafka Streams examples
	git clone –depth 1 -b main \
	https://github.com/garystafford/kstreams-kafka-demo.git

view raw github_projects.sh hosted with ❤ by GitHub

Docker

To make it easier to follow along with the demonstration, we will use Docker Swarm to provision the streaming tools. Alternatively, you could use Kubernetes (e.g., creating a Helm chart) or your preferred CSP or SaaS managed services. Nothing in this demonstration requires you to use a paid service.

The two Docker Swarm stacks are located in the Streaming Synthetic Sales Data Generator project:

1. Streaming Stack — Part 1: Apache Kafka, Apache Zookeeper, Apache Spark, UI for Apache Kafka, and the KStreams application
2. Streaming Stack — Part 2: Apache Kafka, Apache Zookeeper, Apache Flink, Apache Pinot, Apache Superset, UI for Apache Kafka, and Project Jupyter (JupyterLab).*

* the Jupyter container can be used as an alternative to the Spark container for running PySpark jobs (follow the same steps as for Spark, below)

Demonstration #1: Apache Spark

In the first of four demonstrations, we will examine two Apache Spark Structured Streaming jobs, written in PySpark, demonstrating both batch processing (spark_batch_kafka.py) and stream processing (spark_streaming_kafka.py). We will read from a single stream of data from a Kafka topic, demo.purchases, and write to the console.

Deploying the Streaming Stack

To get started, deploy the first streaming Docker Swarm stack containing the Apache Kafka, Apache Zookeeper, Apache Spark, UI for Apache Kafka, and the KStreams application containers.

	# cd into project
	cd streaming-sales-generator/
	# initialize swarm stack – 1x only
	docker swarm init
	# optional: delete previous streaming-stack
	docker stack rm streaming-stack
	# deploy first streaming-stack
	docker stack deploy streaming-stack –compose-file docker/spark-kstreams-stack.yml
	# observe the deployment's progress
	docker stack services streaming-stack

view raw deploy_streaming_stack_1.sh hosted with ❤ by GitHub

The stack will take a few minutes to deploy fully. When complete, there should be a total of six containers running in the stack.

Sales Generator

Before starting the streaming data generator, confirm or modify the configuration/configuration.ini. Three configuration items, in particular, will determine how long the streaming data generator runs and how much data it produces. We will set the timing of transaction events to be generated relatively rapidly for test purposes. We will also set the number of events high enough to give us time to explore the Spark jobs. Using the below settings, the generator should run for an average of approximately 50–60 minutes: (((5 sec + 2 sec)/2)*1000 transactions)/60 sec=~58 min on average. You can run the generator again if necessary or increase the number of transactions.

	[SALES]
	# minimum sales frequency in seconds (debug with 1, typical min. 120)
	min_sale_freq = 2
	# maximum sales frequency in seconds (debug with 3, typical max. 300)
	max_sale_freq = 5
	# number of transactions to generate
	number_of_sales = 1000

view raw configuration.ini hosted with ❤ by GitHub

Start the streaming data generator as a background service:

	# install required python packages (1x)
	python3 -m pip install kafka-python
	cd sales_generator/
	# run in foreground
	python3 ./producer.py
	# better option, run as background process
	nohup python3 ./producer.py &
	# confirm process is running
	ps -u

view raw run_sales_generator.sh hosted with ❤ by GitHub

The streaming data generator will start writing data to three Apache Kafka topics: demo.products, demo.purchases, and demo.inventories. We can view these topics and their messages by logging into the Apache Kafka container and using the Kafka CLI:

	# establish an interactive session with the spark container
	KAFKA_CONTAINER=$(docker container ls –filter name=streaming-stack_kafka.1 –format "{{.ID}}")
	docker exec -it ${KAFKA_CONTAINER} bash
	# set environment variables used by jobs
	export BOOTSTRAP_SERVERS="localhost:9092"
	export TOPIC_PRODUCTS="demo.products"
	export TOPIC_PURCHASES="demo.purchases"
	export TOPIC_INVENTORIES="demo.inventories"
	# list topics
	kafka-topics.sh –list –bootstrap-server $BOOTSTRAP_SERVERS
	# read topics from beginning
	kafka-console-consumer.sh \
	–topic $TOPIC_PRODUCTS –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS
	kafka-console-consumer.sh \
	–topic $TOPIC_PURCHASES –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS
	kafka-console-consumer.sh \
	–topic $TOPIC_INVENTORIES –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS

view raw read_kafka_topics.sh hosted with ❤ by GitHub

Below, we see a few sample messages from the demo.purchases topic:

Alternatively, you can use the UI for Apache Kafka, accessible on port 9080.

Prepare Spark

Next, prepare the Spark container to run the Spark jobs:

	# establish an interactive session with the spark container
	SPARK_CONTAINER=$(docker container ls –filter name=streaming-stack_spark.1 –format "{{.ID}}")
	docker exec -it -u 0 ${SPARK_CONTAINER} bash
	# update and install wget
	apt-get update && apt-get install wget vim -y
	# install required job dependencies
	wget https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.11.1/commons-pool2-2.11.1.jar
	wget https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.3.1/kafka-clients-3.3.1.jar
	wget https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.3.1/spark-sql-kafka-0-10_2.12-3.3.1.jar
	wget https://repo1.maven.org/maven2/org/apache/spark/spark-token-provider-kafka-0-10_2.12/3.3.1/spark-token-provider-kafka-0-10_2.12-3.3.1.jar
	mv *.jar /opt/bitnami/spark/jars/
	exit

view raw prepare_spark_container.sh hosted with ❤ by GitHub

Running the Spark Jobs

Next, copy the jobs from the project to the Spark container, then exec back into the container:

	# copy jobs to spark container
	docker cp apache_spark_examples/ ${SPARK_CONTAINER}:/home/
	# establish an interactive session with the spark container
	docker exec -it ${SPARK_CONTAINER} bash

view raw run_spark_jobs.sh hosted with ❤ by GitHub

Batch Processing with Spark

The first Spark job, spark_batch_kafka.py, aggregates the number of items sold and the total sales for each product, based on existing messages consumed from the demo.purchases topic. We use the PySpark DataFrame class’s read() and write() methods in the first example, reading from Kafka and writing to the console. We could just as easily write the results back to Kafka.

	ds_sales = (
	df_sales.selectExpr("CAST(value AS STRING)")
	.select(F.from_json("value", schema=schema).alias("data"))
	.select("data.*")
	.withColumn("row", F.row_number().over(window))
	.withColumn("quantity", F.sum(F.col("quantity")).over(window_agg))
	.withColumn("sales", F.sum(F.col("total_purchase")).over(window_agg))
	.filter(F.col("row") == 1)
	.drop("row")
	.select(
	"product_id",
	F.format_number("sales", 2).alias("sales"),
	F.format_number("quantity", 0).alias("quantity"),
	)
	.coalesce(1)
	.orderBy(F.regexp_replace("sales", ",", "").cast("float"), ascending=False)
	.write.format("console")
	.option("numRows", 25)
	.option("truncate", False)
	.save()
	)

view raw spark_batch_snippet.py hosted with ❤ by GitHub

The batch processing job sorts the results and outputs the top 25 items by total sales to the console. The job should run to completion and exit successfully.

To run the batch Spark job, use the following commands:

	# set environment variables used by jobs
	export BOOTSTRAP_SERVERS="kafka:29092"
	export TOPIC_PURCHASES="demo.purchases"
	cd /home/apache_spark_examples/
	# run batch processing job
	spark-submit spark_batch_kafka.py

view raw run_spark_batch.sh hosted with ❤ by GitHub

Stream Processing with Spark

The stream processing Spark job, spark_streaming_kafka.py, also aggregates the number of items sold and the total sales for each item, based on messages consumed from the demo.purchases topic. However, as shown in the code snippet below, this job continuously aggregates the stream of data from Kafka, displaying the top ten product totals within an arbitrary ten-minute sliding window, with a five-minute overlap, and updates output every minute to the console. We use the PySpark DataFrame class’s readStream() and writeStream() methods as opposed to the batch-oriented read() and write() methods in the first example.

	ds_sales = (
	df_sales.selectExpr("CAST(value AS STRING)")
	.select(F.from_json("value", schema=schema).alias("data"))
	.select("data.*")
	.withWatermark("transaction_time", "10 minutes")
	.groupBy("product_id", F.window("transaction_time", "10 minutes", "5 minutes"))
	.agg(F.sum("total_purchase"), F.sum("quantity"))
	.orderBy(F.col("window").desc(), F.col("sum(total_purchase)").desc())
	.select(
	"product_id",
	F.format_number("sum(total_purchase)", 2).alias("sales"),
	F.format_number("sum(quantity)", 0).alias("drinks"),
	"window.start",
	"window.end",
	)
	.coalesce(1)
	.writeStream.queryName("streaming_to_console")
	.trigger(processingTime="1 minute")
	.outputMode("complete")
	.format("console")
	.option("numRows", 10)
	.option("truncate", False)
	.start()
	)
	ds_sales.awaitTermination()

view raw spark_streaming_snippet.py hosted with ❤ by GitHub

Shorter event-time windows are easier for demonstrations — in Production, hourly, daily, weekly, or monthly windows are more typical for sales analysis.

To run the stream processing Spark job, use the following commands:

	# run stream processing job
	spark-submit spark_streaming_kafka.py

view raw run_spark_steaming_job.sh hosted with ❤ by GitHub

We could just as easily calculate running totals for the stream of sales data versus aggregations over a sliding event-time window (example job included in project).

Be sure to kill the stream processing Spark jobs when you are done, or they will continue to run, awaiting more data.

Demonstration #2: Apache Kafka Streams

Next, we will examine Apache Kafka Streams (aka KStreams). For this part of the post, we will also use the second of the three GitHub repository projects, kstreams-kafka-demo. The project contains a KStreams application written in Java that performs stream processing and incremental aggregation.

KStreams Application

The KStreams application continuously consumes the stream of messages from the demo.purchases Kafka topic (source) using an instance of the StreamBuilder() class. It then aggregates the number of items sold and the total sales for each item, maintaining running totals, which are then streamed to a new demo.running.totals topic (sink). All of this using an instance of the KafkaStreams() Kafka client class.

	private static void kStreamPipeline(Properties props) {
	System.out.println("Starting…");
	Properties kafkaStreamsProps = new Properties();
	kafkaStreamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, props.getProperty("APPLICATION_ID"));
	kafkaStreamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty("BOOTSTRAP_SERVERS"));
	kafkaStreamsProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, props.getProperty("AUTO_OFFSET_RESET_CONFIG"));
	kafkaStreamsProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, props.getProperty("COMMIT_INTERVAL_MS_CONFIG"));
	StreamsBuilder builder = new StreamsBuilder();
	builder
	.stream(props.getProperty("INPUT_TOPIC"), Consumed.with(Serdes.Void(), CustomSerdes.Purchase()))
	.peek((unused, purchase) -> System.out.println(purchase.toString()))
	.flatMap((KeyValueMapper<Void, Purchase, Iterable<KeyValue<String, Total>>>) (unused, purchase) -> {
	List<KeyValue<String, Total>> result = new ArrayList<>();
	result.add(new KeyValue<>(purchase.getProductId(), new Total(
	purchase.getTransactionTime(),
	purchase.getProductId(),
	1,
	purchase.getQuantity(),
	purchase.getTotalPurchase()
	)));
	return result;
	})
	.groupByKey(Grouped.with(Serdes.String(), CustomSerdes.Total()))
	.reduce((total1, total2) -> {
	total2.setTransactions(total1.getTransactions() + total2.getTransactions());
	total2.setQuantities(total1.getQuantities() + total2.getQuantities());
	total2.setSales(total1.getSales().add(total2.getSales()));
	return total2;
	})
	.toStream()
	.peek((productId, total) -> System.out.println(total.toString()))
	.to(props.getProperty("OUTPUT_TOPIC"), Produced.with(Serdes.String(), CustomSerdes.Total()));
	KafkaStreams streams = new KafkaStreams(builder.build(), kafkaStreamsProps);
	streams.start();
	System.out.println("Running…");
	}

view raw kstream_snippet.java hosted with ❤ by GitHub

Running the Application

We have at least three choices to run the KStreams application for this demonstration: 1) running locally from our IDE, 2) a compiled JAR run locally from the command line, or 3) a compiled JAR copied into a Docker image, which is deployed as part of the Swarm stack. You can choose any of the options.

	# set java version (v17 is latest compatible version with kstreams)
	JAVA_HOME=~/Library/Java/JavaVirtualMachines/corretto-17.0.5/Contents/Home
	$JAVA_HOME/bin/java -version
	# compile to uber jar
	./gradlew clean shadowJar
	# run the streaming application
	$JAVA_HOME/bin/java -jar build/libs/kstreams-kafka-demo-1.0.0-all.jar

view raw run_kstreams_app.sh hosted with ❤ by GitHub

Compiling and running the KStreams application locally

We will continue to use the same streaming Docker Swarm stack used for the Apache Spark demonstration. I have already compiled a single uber JAR file using OpenJDK 17 and Gradle from the project’s source code. I then created and published a Docker image, which is already part of the running stack.

	FROM amazoncorretto:17.0.5
	COPY build/libs/kstreams-kafka-demo-1.1.0-all.jar /tmp/kstreams-app.jar
	CMD ["java", "-jar", "/tmp/kstreams-app.jar"]

view raw Dockerfile hosted with ❤ by GitHub

Since we ran the sales generator earlier for the Spark demonstration, there is existing data in the demo.purchases topic. Re-run the sales generator (nohup python3 ./producer.py &) to generate a new stream of data. View the results of the KStreams application, which has been running since the stack was deployed using the Kafka CLI or UI for Apache Kafka:

	# terminal 1: establish an interactive session with the kstreams app container
	KSTREAMS_CONTAINER=$(docker container ls –filter name=streaming-stack_kstreams.1 –format "{{.ID}}")
	docker logs ${KSTREAMS_CONTAINER} –follow
	# terminal 2: establish an interactive session with the kafka container
	KAFKA_CONTAINER=$(docker container ls –filter name=streaming-stack_kafka.1 –format "{{.ID}}")
	docker exec -it ${KAFKA_CONTAINER} bash
	# set environment variables used by jobs
	export BOOTSTRAP_SERVERS="localhost:9092"
	export INPUT_TOPIC="demo.purchases"
	export OUTPUT_TOPIC="demo.running.totals"
	# read topics from beginning
	kafka-console-consumer.sh \
	–topic $INPUT_TOPIC –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS
	kafka-console-consumer.sh \
	–topic $OUTPUT_TOPIC –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS

view raw kstreams_app_output.sh hosted with ❤ by GitHub

Below, in the top terminal window, we see the output from the KStreams application. Using KStream’s peek() method, the application outputs Purchase and Total instances to the console as they are processed and written to Kafka. In the lower terminal window, we see new messages being published as a continuous stream to output topic, demo.running.totals.

Part Two

In part two of this two-part post, we continue our exploration of the four popular open-source stream processing projects. We will cover Apache Flink and Apache Pinot. In addition, we will incorporate Apache Superset into the demonstration, building a real-time dashboard to visualize the results of our stream processing.

---

This blog represents my viewpoints and not of my employer, Amazon Web Services (AWS). All product names, logos, and brands are the property of their respective owners. All diagrams and illustrations are the property of the author unless otherwise noted.