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.”

This two-part post series and forthcoming video explore four popular open-source software (OSS) stream processing projects: Apache Spark Structured Streaming, Apache Kafka Streams, Apache Flink, and Apache Pinot.

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.

Part Two

We will continue our exploration in part two of this two-part post, covering Apache Flink and Apache Pinot. In addition, we will incorporate Apache Superset into the demonstration to visualize the real-time results of our stream processing pipelines as a dashboard.

Demonstration #3: Apache Flink

In the third demonstration of four, we will examine Apache Flink. For this part of the post, we will also use the third of the three GitHub repository projects, flink-kafka-demo. The project contains a Flink application written in Java, which performs stream processing, incremental aggregation, and multi-stream joins.

New Streaming Stack

To get started, we need to replace the first streaming Docker Swarm stack, deployed in part one, with the second streaming Docker Swarm stack. The second stack contains Apache Kafka, Apache Zookeeper, Apache Flink, Apache Pinot, Apache Superset, UI for Apache Kafka, and Project Jupyter (JupyterLab).

https://programmaticponderings.wordpress.com/media/601efca17604c3a467a4200e93d7d3ff

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

Flink Application

The Flink application has two entry classes. The first class, RunningTotals, performs an identical aggregation function as the previous KStreams demo.

	public static void flinkKafkaPipeline(Properties prop) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	// assumes PLAINTEXT authentication
	KafkaSource<Purchase> source = KafkaSource.<Purchase>builder()
	.setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
	.setTopics(prop.getProperty("PURCHASES_TOPIC"))
	.setGroupId("flink_reduce_demo")
	.setStartingOffsets(OffsetsInitializer.earliest())
	.setValueOnlyDeserializer(new PurchaseDeserializationSchema())
	.build();
	DataStream<Purchase> purchases = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
	DataStream<RunningTotal> runningTotals = purchases
	.flatMap((FlatMapFunction<Purchase, RunningTotal>) (purchase, out) -> out.collect(
	new RunningTotal(
	purchase.getTransactionTime(),
	purchase.getProductId(),
	1,
	purchase.getQuantity(),
	purchase.getTotalPurchase()
	))
	).returns(RunningTotal.class)
	.keyBy(RunningTotal::getProductId)
	.reduce((runningTotal1, runningTotal2) -> {
	runningTotal2.setTransactions(runningTotal1.getTransactions() + runningTotal2.getTransactions());
	runningTotal2.setQuantities(runningTotal1.getQuantities() + runningTotal2.getQuantities());
	runningTotal2.setSales(runningTotal1.getSales().add(runningTotal2.getSales()));
	return runningTotal2;
	});
	KafkaSink<RunningTotal> sink = KafkaSink.<RunningTotal>builder()
	.setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
	.setRecordSerializer(KafkaRecordSerializationSchema.builder()
	.setTopic(prop.getProperty("RUNNING_TOTALS_TOPIC"))
	.setValueSerializationSchema(new RunningTotalSerializationSchema())
	.build()
	).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
	.build();
	runningTotals.sinkTo(sink);
	env.execute("Flink Running Totals Demo");
	}

view raw RunningTotals.java hosted with ❤ by GitHub

The second class, JoinStreams, joins the stream of data from the demo.purchases topic and the demo.products topic, processing and combining them, in real-time, into an enriched transaction and publishing the results to a new topic, demo.purchases.enriched.

	public static void flinkKafkaPipeline(Properties prop) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
	// assumes PLAINTEXT authentication
	KafkaSource<Product> productSource = KafkaSource.<Product>builder()
	.setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
	.setTopics(prop.getProperty("PRODUCTS_TOPIC"))
	.setGroupId("flink_join_demo")
	.setStartingOffsets(OffsetsInitializer.earliest())
	.setValueOnlyDeserializer(new ProductDeserializationSchema())
	.build();
	DataStream<Product> productsStream = env.fromSource(
	productSource, WatermarkStrategy.noWatermarks(), "Kafka Products Source");
	tableEnv.createTemporaryView("products", productsStream);
	KafkaSource<Purchase> purchasesSource = KafkaSource.<Purchase>builder()
	.setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
	.setTopics(prop.getProperty("PURCHASES_TOPIC"))
	.setGroupId("flink_join_demo")
	.setStartingOffsets(OffsetsInitializer.earliest())
	.setValueOnlyDeserializer(new PurchaseDeserializationSchema())
	.build();
	DataStream<Purchase> purchasesStream = env.fromSource(
	purchasesSource, WatermarkStrategy.noWatermarks(), "Kafka Purchases Source");
	tableEnv.createTemporaryView("purchases", purchasesStream);
	Table result =
	tableEnv.sqlQuery(
	"SELECT " +
	"purchases.transactionTime, " +
	"TO_TIMESTAMP(purchases.transactionTime), " +
	"purchases.transactionId, " +
	"purchases.productId, " +
	"products.category, " +
	"products.item, " +
	"products.size, " +
	"products.cogs, " +
	"products.price, " +
	"products.containsFruit, " +
	"products.containsVeggies, " +
	"products.containsNuts, " +
	"products.containsCaffeine, " +
	"purchases.price, " +
	"purchases.quantity, " +
	"purchases.isMember, " +
	"purchases.memberDiscount, " +
	"purchases.addSupplements, " +
	"purchases.supplementPrice, " +
	"purchases.totalPurchase " +
	"FROM " +
	"products " +
	"JOIN purchases " +
	"ON products.productId = purchases.productId"
	);
	DataStream<PurchaseEnriched> purchasesEnrichedTable = tableEnv.toDataStream(result,
	PurchaseEnriched.class);
	KafkaSink<PurchaseEnriched> sink = KafkaSink.<PurchaseEnriched>builder()
	.setBootstrapServers(prop.getProperty("BOOTSTRAP_SERVERS"))
	.setRecordSerializer(KafkaRecordSerializationSchema.builder()
	.setTopic(prop.getProperty("PURCHASES_ENRICHED_TOPIC"))
	.setValueSerializationSchema(new PurchaseEnrichedSerializationSchema())
	.build()
	).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
	.build();
	purchasesEnrichedTable.sinkTo(sink);
	env.execute("Flink Streaming Join Demo");
	}

view raw JoinStreams.java hosted with ❤ by GitHub

The resulting enriched purchases messages look similar to the following:

	{
	"transaction_time": "2022-09-25 01:58:11.714838",
	"transaction_id": "4068565608708439642",
	"product_id": "CS08",
	"product_category": "Classic Smoothies",
	"product_name": "Rockin’ Raspberry",
	"product_size": "24 oz.",
	"product_cogs": 1.5,
	"product_price": 4.99,
	"contains_fruit": true,
	"contains_veggies": false,
	"contains_nuts": false,
	"contains_caffeine": false,
	"purchase_price": 4.99,
	"purchase_quantity": 2,
	"is_member": false,
	"member_discount": 0,
	"add_supplements": false,
	"supplement_price": 0,
	"total_purchase": 9.98
	}

view raw enriched_purchase.json hosted with ❤ by GitHub

Running the Flink Job

To run the Flink application, we must first compile it into an uber JAR.

	# set java version (v11 is latest compatible version with Apache Flink)
	JAVA_HOME=~/Library/Java/JavaVirtualMachines/corretto-11.0.16.1/Contents/Home
	# compile to uber jar
	./gradlew clean shadowJar

view raw build_flink_app.sh hosted with ❤ by GitHub

We can copy the JAR into the Flink container or upload it through the Apache Flink Dashboard, a browser-based UI. For this demonstration, we will upload it through the Apache Flink Dashboard, accessible on port 8081.

The project’s build.gradle file has preset the Main class (Flink’s Entry class) to org.example.JoinStreams. Optionally, to run the Running Totals demo, we could change the build.gradle file and recompile, or simply change Flink’s Entry class to org.example.RunningTotals.

Before running the Flink job, restart the sales generator in the background (nohup python3 ./producer.py &) to generate a new stream of data. Then start the Flink job.

To confirm the Flink application is running, we can check the contents of the new demo.purchases.enriched topic using the Kafka CLI.

	# 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 OUTPUT_TOPIC="demo.purchases.enriched";
	# read topics from beginning
	kafka-console-consumer.sh \
	–topic $OUTPUT_TOPIC –from-beginning \
	–bootstrap-server $BOOTSTRAP_SERVERS

view raw consume_flink_job.sh hosted with ❤ by GitHub

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

Demonstration #4: Apache Pinot

In the fourth and final demonstration, we will explore Apache Pinot. First, we will query the unbounded data streams from Apache Kafka, generated by both the sales generator and the Apache Flink application, using SQL. Then, we build a real-time dashboard in Apache Superset, with Apache Pinot as our datasource.

Creating Tables

According to the Apache Pinot documentation, “a table is a logical abstraction that represents a collection of related data. It is composed of columns and rows (known as documents in Pinot).” There are three types of Pinot tables: Offline, Realtime, and Hybrid. For this demonstration, we will create three Realtime tables. Realtime tables ingest data from streams — in our case, Kafka — and build segments from the consumed data. Further, according to the documentation, “each table in Pinot is associated with a Schema. A schema defines what fields are present in the table along with the data types. The schema is stored in Zookeeper, along with the table configuration.”

Below, we see the schema and config for one of the three Realtime tables, purchasesEnriched. Note how the columns are divided into three categories: Dimension, Metric, and DateTime.

	{
	"schemaName": "purchasesEnriched",
	"dimensionFieldSpecs": [
	{
	"name": "transaction_id",
	"dataType": "STRING"
	},
	{
	"name": "product_id",
	"dataType": "STRING"
	},
	{
	"name": "product_category",
	"dataType": "STRING"
	},
	{
	"name": "product_name",
	"dataType": "STRING"
	},
	{
	"name": "product_size",
	"dataType": "STRING"
	},
	{
	"name": "product_cogs",
	"dataType": "FLOAT"
	},
	{
	"name": "product_price",
	"dataType": "FLOAT"
	},
	{
	"name": "contains_fruit",
	"dataType": "BOOLEAN"
	},
	{
	"name": "contains_veggies",
	"dataType": "BOOLEAN"
	},
	{
	"name": "contains_nuts",
	"dataType": "BOOLEAN"
	},
	{
	"name": "contains_caffeine",
	"dataType": "BOOLEAN"
	},
	{
	"name": "purchase_price",
	"dataType": "FLOAT"
	},
	{
	"name": "is_member",
	"dataType": "BOOLEAN"
	},
	{
	"name": "member_discount",
	"dataType": "FLOAT"
	},
	{
	"name": "add_supplements",
	"dataType": "BOOLEAN"
	},
	{
	"name": "supplement_price",
	"dataType": "FLOAT"
	}
	],
	"metricFieldSpecs": [
	{
	"name": "purchase_quantity",
	"dataType": "INT"
	},
	{
	"name": "total_purchase",
	"dataType": "FLOAT"
	}
	],
	"dateTimeFieldSpecs": [
	{
	"name": "transaction_time",
	"dataType": "TIMESTAMP",
	"format": "1:MILLISECONDS:SIMPLE_DATE_FORMAT:yyyy-MM-dd HH:mm:ss.SSSSSS",
	"granularity": "1:MILLISECONDS"
	}
	]
	}

view raw purchases-enriched-schema.json hosted with ❤ by GitHub

	{
	"tableName": "purchasesEnriched",
	"tableType": "REALTIME",
	"segmentsConfig": {
	"timeColumnName": "transaction_time",
	"timeType": "MILLISECONDS",
	"schemaName": "purchasesEnriched",
	"replicasPerPartition": "1"
	},
	"tenants": {},
	"tableIndexConfig": {
	"loadMode": "MMAP",
	"streamConfigs": {
	"streamType": "kafka",
	"stream.kafka.consumer.type": "lowlevel",
	"stream.kafka.topic.name": "demo.purchases.enriched",
	"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
	"stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
	"stream.kafka.broker.list": "kafka:29092",
	"realtime.segment.flush.threshold.time": "3600000",
	"realtime.segment.flush.threshold.rows": "50000",
	"stream.kafka.consumer.prop.auto.offset.reset": "smallest"
	}
	},
	"metadata": {}
	}

view raw purchases-enriched-config.json hosted with ❤ by GitHub

To begin, copy the three Pinot Realtime table schemas and configurations from the streaming-sales-generator GitHub project into the Apache Pinot Controller container. Next, use a docker exec command to call the Pinot Command Line Interface’s (CLI) AddTable command to create the three tables: products, purchases, and purchasesEnriched.

	# copy pinot table schema and config files to pinot controller
	CONTROLLER_CONTAINER=$(docker container ls –filter name=streaming-stack_pinot-controller.1 –format "{{.ID}}")
	cd ~/streaming-sales-generator/apache_pinot_examples
	docker cp configs_schemas/ ${CONTROLLER_CONTAINER}:/tmp/
	# create three tables
	docker exec -it ${CONTROLLER_CONTAINER} \
	bin/pinot-admin.sh AddTable \
	-tableConfigFile /tmp/configs_schemas/purchases-config.json \
	-schemaFile /tmp/configs_schemas/purchases-schema.json -exec
	docker exec -it ${CONTROLLER_CONTAINER} \
	bin/pinot-admin.sh AddTable \
	-tableConfigFile /tmp/configs_schemas/products-config.json \
	-schemaFile /tmp/configs_schemas/products-schema.json -exec
	docker exec -it ${CONTROLLER_CONTAINER} \
	bin/pinot-admin.sh AddTable \
	-tableConfigFile /tmp/configs_schemas/purchases-enriched-config.json \
	-schemaFile /tmp/configs_schemas/purchases-enriched-schema.json -exec

view raw create_new_tables.sh hosted with ❤ by GitHub

To confirm the three tables were created correctly, use the Apache Pinot Data Explorer accessible on port 9000. Use the Tables tab in the Cluster Manager.

We can further inspect and edit the table’s config and schema from the Tables tab in the Cluster Manager.

The three tables are configured to read the unbounded stream of data from the corresponding Kafka topics: demo.products, demo.purchases, and demo.purchases.enriched.

Querying with Pinot

We can use Pinot’s Query Console to query the Realtime tables using SQL. According to the documentation, “Pinot provides a SQL interface for querying. It uses the [Apache] Calcite SQL parser to parse queries and uses MYSQL_ANSI dialect.”

With the generator still running, re-query the purchases table in the Query Console (select count(*) from purchases). You should notice the document count increasing each time you re-run the query since new messages are published to the demo.purchases topic by the sales generator.

If you do not observe the count increasing, ensure the sales generator and Flink enrichment job are running.

Table Joins?

It might seem logical to want to replicate the same multi-stream join we performed with Apache Flink in part three of the demonstration on the demo.products and demo.purchases topics. Further, we might presume to join the products and purchases realtime tables by writing a SQL statement in Pinot’s Query Console. However, according to the documentation, at the time of this post, version 0.11.0 of Pinot did not [currently] support joins or nested subqueries.

This current join limitation is why we created the Realtime table, purchasesEnriched, allowing us to query Flink’s real-time results in the demo.purchases.enriched topic. We will use both Flink and Pinot as part of our stream processing pipeline, taking advantage of each tool’s individual strengths and capabilities.

Note, according to the documentation for the latest release of Pinot on the main branch, “the latest Pinot multi-stage supports inner join, left-outer, semi-join, and nested queries out of the box. It is optimized for in-memory process and latency.” For more information on joins as part of Pinot’s new multi-stage query execution engine, read the documentation, Multi-Stage Query Engine.

Aggregations

We can perform real-time aggregations using Pinot’s rich SQL query interface. For example, like previously with Spark and Flink, we can calculate running totals for the number of items sold and the total sales for each product in real time.

We can do the same with the purchasesEnriched table, which will use the continuous stream of enriched transaction data from our Apache Flink application. With the purchasesEnriched table, we can add the product name and product category for richer results. Each time we run the query, we get real-time results based on the running sales generator and Flink enrichment job.

Query Options and Indexing

Note the reference to the Star-Tree index at the start of the SQL query shown above. Pinot provides several query options, including useStarTree (true by default).

Multiple indexing techniques are available in Pinot, including Forward Index, Inverted Index, Star-tree Index, Bloom Filter, and Range Index, among others. Each has advantages in different query scenarios. According to the documentation, by default, Pinot creates a dictionary-encoded forward index for each column.

SQL Examples

Here are a few examples of SQL queries you can try in Pinot’s Query Console:

	— products
	SELECT
	COUNT(product_id) AS product_count,
	AVG(price) AS avg_price,
	AVG(cogs) AS avg_cogs,
	AVG(price) – AVG(cogs) AS avg_gross_profit
	FROM
	products;
	— purchases
	SELECT
	product_id,
	SUMPRECISION(quantity, 10, 0) AS quantity,
	SUMPRECISION(total_purchase, 10, 2) AS sales
	FROM
	purchases
	GROUP BY
	product_id
	ORDER BY
	sales DESC;
	— purchasesEnriched
	SELECT
	product_id,
	product_name,
	product_category,
	SUMPRECISION(purchase_quantity, 10, 0) AS quantity,
	SUMPRECISION(total_purchase, 10, 2) AS sales
	FROM
	purchasesEnriched
	GROUP BY
	product_id,
	product_name,
	product_category
	ORDER BY
	sales DESC;

view raw pinot_sql_example.sql hosted with ❤ by GitHub

Troubleshooting Pinot

If have issues with creating the tables or querying the real-time data, you can start by reviewing the Apache Pinot logs:

	CONTROLLER_CONTAINER=$(docker container ls –filter name=streaming-stack_pinot-controller.1 –format "{{.ID}}")
	docker exec -it ${CONTROLLER_CONTAINER} cat logs/pinot-all.log

view raw pinot_logs.sh hosted with ❤ by GitHub

Real-time Dashboards with Apache Superset

To display the real-time stream of data produced results of our Apache Flink stream processing job and made queriable by Apache Pinot, we can use Apache Superset. Superset positions itself as “a modern data exploration and visualization platform.” Superset allows users “to explore and visualize their data, from simple line charts to highly detailed geospatial charts.”

According to the documentation, “Superset requires a Python DB-API database driver and a SQLAlchemy dialect to be installed for each datastore you want to connect to.” In the case of Apache Pinot, we can use pinotdb as the Python DB-API and SQLAlchemy dialect for Pinot. Since the existing Superset Docker container does not have pinotdb installed, I have built and published a Docker Image with the driver and deployed it as part of the second streaming stack of containers.

	# Custom Superset build to add apache pinot driver
	# Gary A. Stafford (2022-09-25)
	# Updated: 2022-12-18
	FROM apache/superset:66138b0ca0b82a94404e058f0cc55517b2240069
	# Switching to root to install the required packages
	USER root
	# Find which driver you need based on the analytics database:
	# https://superset.apache.org/docs/databases/installing-database-drivers
	RUN pip install mysqlclient psycopg2-binary pinotdb
	# Switching back to using the `superset` user
	USER superset

view raw Dockerfile hosted with ❤ by GitHub

First, we much configure the Superset container instance. These instructions are documented as part of the Superset Docker Image repository.

	# establish an interactive session with the superset container
	SUPERSET_CONTAINER=$(docker container ls –filter name=streaming-stack_superset.1 –format "{{.ID}}")
	# initialize superset (see superset documentation)
	docker exec -it ${SUPERSET_CONTAINER} \
	superset fab create-admin \
	–username admin \
	–firstname Superset \
	–lastname Admin \
	–email admin@superset.com \
	–password sUp3rS3cREtPa55w0rD1
	docker exec -it ${SUPERSET_CONTAINER} superset db upgrade
	docker exec -it ${SUPERSET_CONTAINER} superset init

view raw initialize_superset.sh hosted with ❤ by GitHub

Once the configuration is complete, we can log into the Superset web-browser-based UI accessible on port 8088.

Pinot Database Connection and Dataset

Next, to connect to Pinot from Superset, we need to create a Database Connection and a Dataset.

The SQLAlchemy URI is shown below. Input the URI, test your connection (‘Test Connection’), make sure it succeeds, then hit ‘Connect’.

pinot+http://pinot-broker:8099/query?controller=http://pinot-controller:9000

view raw pinot_superset_conn.txt hosted with ❤ by GitHub

Next, create a Dataset that references the purchasesEnriched Pinot table.

Modify the dataset’s transaction_time column. Check the is_temporal and Default datetime options. Lastly, define the DateTime format as epoch_ms.

Building a Real-time Dashboard

Using the new dataset, which connects Superset to the purchasesEnriched Pinot table, we can construct individual charts to be placed on a dashboard. Build a few charts to include on your dashboard.

Create a new Superset dashboard and add the charts and other elements, such as headlines, dividers, and tabs.

We can apply a refresh interval to the dashboard to continuously query Pinot and visualize the results in near real-time.

Conclusion

In this two-part post series, we were introduced to stream processing. We explored four popular open-source stream processing projects: Apache Spark Structured Streaming, Apache Kafka Streams, Apache Flink, and Apache Pinot. Next, we learned how we could solve similar stream processing and streaming analytics challenges using different streaming technologies. Lastly, we saw how these technologies, such as Kafka, Flink, Pinot, and Superset, could be integrated to create effective stream processing pipelines.

---

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.