Streaming SQL Cheat Sheet for Apache Kafka

What is Lenses?

Lenses is developer experience for building and operating streaming applications on open-source technologies. Deployed as a container or as a JVM application, Lenses.io works with any Kafka & Kubernetes environment including managed and cloud services.

Lenses Streaming SQL

Streaming SQL allows anyone to build stream processing applications with SQL. Queries react to data as soon as it is available in Apache Kafka. They can reshape your data, aggregate it based on any field or time window, or enrich it with other streaming data. The results are pushed back to Kafka topics so that downstream applications and processes can consume them. Pipelines are deployed and scaled on your own Kubernetes environment or managed cloud solutions.

Try out Streaming SQL

Stateful and Stateless processing

Streams (Stateless) and tables (Stateful) have different semantics and use-cases, but are strongly related nonetheless.

This relationship is known as stream-table duality. Every stream can be interpreted as a table, and similarly a table can be interpreted as a stream.

SELECT STREAM *
FROM website_traffic

SELECT TABLE *
FROM customers

Streaming SQL example use cases

Supported formats

Lenses.io supports reading and writing data from/to Kafka topics in different serialisation formats.

First basic example

SET defaults.topic.autocreate=true;

INSERT INTO speeding_cars
SELECT STREAM
	car_speed as speed
	 , car_name
WHERE car_speed > 100
FROM car_data;

The above example will auto-create any necessary topics (in this case, “speeding_cars” topic) and populate it with the value of car_speed renamed as speed and car_name from the car_data topic only for events where the car_speed is greater than 100.

Kafka Record/Message

INSERT INTO outputTopic 
SELECT STREAM _key as customerId
FROM customers

INSERT INTO outputTopic
SELECT STREAM _value.firstName 
FROM customers

Projections

A projection represents the ability to project a given input value onto a target location (key or value) in an output record. Projections are the main building block of SELECT statements.

INSERT INTO target-topic
SELECT STREAM
	CONCAT('a', 'b') AS result1
	, (1 + field1) AS _key.a
	, _key.field2 AS result3
	, CASE
		WHEN field3 = 'Robert' THEN 'It`s bobby'
		WHEN field3 ='William' THEN 'It`s willy'
		ELSE 'Unknown'
		END AS who_is_it
FROM input-topic;

In the above example, result1, _key.a, result3 and who_is_itare all fields outputted into target-topic.

Operators & Expressions

WHERE 1 + _value.field1 > 5 
AND value.field2 < 10

WHERE 1 + _value.field1 
> 5

WHERE _value.field1 
> _value.field2

WHERE _key !=
LENGTH(_value.field1)

WHERE _value.field1 
NOT LIKE “%foo%”

CASE WHEN field3 = 'Robert' 
THEN 'Its bobby' 
WHEN field3 = 'William'
THEN 'Its willy'
ELSE 'Unknown'
END AS who_is_it

Examples

Select move_number and date_released from transformer_movies_xml topic and put in as AVRO format in transformer_movies_avro topic. Make movie_number as the _key and only select movies where movie_number is less than 300.

INSERT INTO transformer_movies_avro
	STORE KEY AS AVRO
	VALUE AS AVRO
	SELECT STREAM movie_number AS _key
	, date_released
	FROM transformer_movies_xml
	WHERE movie_number > 0 AND
	movie_number < 300

Calculate the rolling average KW for each customer_id for every 10 minute period every 5 minutes. Populate intoelectricity_event_avg stream

INSERT INTO electricity_events_avg
	SELECT STREAM customer_id
	, AVG (KW) AS KW
	FROM electricity_events
	WINDOW BY HOP 10m, 5m
	GROUP BY customer_id

Populate the ship_speeds_with_names topic with the speed field from fast_vessel_processor topic joined with the ship_names topic using the MMSI value. The ship_names MMSI value is stored as a STRING whereas the fast_vessel_processor is stored as a LONG. The ship_names also doesn't have a key so needs to be rekeyed (ship_names_rekeyed) in order to build a state table (ship_names_state) since states cannot have NULL keys.

WITH ship_names_rekeyed AS
	(SELECT STREAM ship_names.mmsi AS _key
	FROM ship_names);
	WITH ship_names_state AS
	(SELECT TABLE *
	FROM ship_names_rekeyed);
	INSERT INTO ship_speeds_with_names
	SELECT STREAM fast_vessel_processor._key AS mmsikey
	, fast_vessel_processor.Speed
	, ship_names.owner
	, ship_names.name
	FROM fast_vessel_processor
	INNER JOIN ship_names_state ON fast_vessel_processor._key.MMSI
	= CAST(ship_names_state._key AS LONG);

Reshape the car_speed_events stream by setting sensor_id andevent_time as keys and nesting mph and a calculated kmph values for speed object in the value field.

SELECT STREAM sensor.id AS _key.sensor_id
	, event_time AS _key.event_time
	, car_id
	, speedMph AS speed.mph
	, speedMph * 1.60934 AS speed.kmph
	FROM car_speed_events;

Changing Storage Format

At times, it is useful to control the resulting Key and/or Value storage of the output topic. If the input is Json, the output for the streaming computation can be set to Avro.

The syntax is the following:

INSERT INTO <target topic>
	STORE 
	KEY AS <format>
	VALUE AS <format>
...

JOINS

Join Types

INSERT INTO result 
SELECT STREAM customersTable.name
, ordersStream.item 
FROM ordersStream 
JOIN customersTable
ON customersTable.id =
ordersStream.customer_id;

INSERT INTO result 
SELECT STREAM customersTable.name
, ordersStream.item 
FROM ordersStream 
LEFT JOIN customersTable
ON customersTable.id =
ordersStream.customer_id;

INSERT INTO result 
SELECT TABLE customersTable.name
, ordersStream.item 
FROM customersTable 
RIGHT JOIN ordersStream
ON customersTable.id =
ordersStream.customer_id;

INSERT INTO result 
SELECT TABLE customersStream.name
, ordersStream.item 
FROM ordersStream 
OUTER JOIN customersStream
ON customersTable.id =
ordersStream.customer_id;

Join Match Expressions

if no ON expression is provided, the join will be evaluated based on the equality of the _key facet

INSERT INTO RESULT 
SELECT TABLE customersStream.name
, ordersStream.item 
FROM ordersStream 
OUTER JOIN customersStream
ON substr(customersTable.name, 5) =
ordersStream.customerName;

INSERT INTO RESULT 
SELECT TABLE cars_table.cars_name
, car_speeds.speed 
FROM cars_table 
OUTER JOIN car_speeds
ON car_speeds.speed > 100

INSERT INTO RESULT 
SELECT TABLE customersStream.name
, customersTable.country
, ordersStream.customerId
, FROM ordersStream 
OUTER JOIN customersStream
ON customersTable.country = "USA"
AND customersTable.id =
ordersStream.customerId

Join Compatibility

Stream-to-Stream Windowing with WITHIN

When two streams are joined Lenses needs to know how far away in the past and in the future to look for a matching record.

SELECT STREAM
	customers.name
	, orders.item
FROM
	customers LEFT JOIN orders WITHIN 5s
		ON customers.id = orders.customer_id
WITHIN 5s;

The above example means that an event with an orders.item may be generated with a null value for customers.name if a matching customers.id event has not been generated within a 5 second +/- window.

Time Windows

Supported time descriptors

Windowing types

Hopping Window	WINDOW BY HOP <duration_time>,<hop_i nterval> Fixed size and overlapping windows. The same event can overlap into multiple windows.
Tumbling window	WINDOW BY TUMBLE <duration_time> Duration and hop interval are equal. An event can only appear in one window.
Session window	WINDOW BY SESSION <inactivity_interval> Defined by a period of activity separated by a specified gap of inactivity at which point current session closes.
Grace period	WINDOW BY <windowing type> GRACE BY <grace time> For a window to accept late-arriving records a grace period can be defined. If a record falls within a window and it arrived before the end of the grace time then the record will be processed and the aggregations or joins will update. If not, the record is discarded.

WITH for better SQL structure

Use `WITH` to break the code in reusable blocks:

WITH countriesStream AS (
	SELECT STREAM *
	FROM countries
);

WITH merchantsStream AS (
	SELECT STREAM *
	FROM merchants
);

WITH merchantsWithCountryInfoStream AS (
	SELECT STREAM
		m._key AS l_key
		, CONCAT(surname, ', ', name) AS fullname
		, address.country
		, language
		, platform
	FROM merchantsStream AS m JOIN countriesStream AS c
			ON m.address.country = c._key
	WITHIN 1h
);

WITH merchantsCorrectKey AS (
	SELECT STREAM
		l_key AS _key
		, fullname
		, country
		, language
		, platform
	FROM merchantsWithCountryInfoStream 
);

INSERT INTO currentMerchants
SELECT STREAM *
FROM merchantsCorrectKey;

INSERT INTO merchantsPerPlatform
SELECT TABLE
	COUNT(*) AS merchants
FROM merchantsCorrectKey
GROUP BY platform;

Functions

String functions

Obfuscation functions

Numeric Functions

Date Functions

Utility Functions

Flags

Allows to customize the behaviour of the application (including error handling). Must be declared at the beginning of the SQL statement.