ksqlDB processes data in realtime, and you can also import and export data straight from ksqlDB from popular data sources and end systems in the cloud.
This tutorial shows you how to run the recipe in one of two ways: using connector(s) to any supported data source or using ksqlDB’s INSERT INTO functionality to mock the data.
If you cannot connect to a real data source with properly formatted data, or if you just want to execute this tutorial without external dependencies, no worries! Remove the CREATE SOURCE CONNECTOR commands and insert mock data into the streams.
This application compares merchant and mobile user geolocation data to produce user proximity alerts. Initially, merchant data is sourced from a database and contains a Geohash value per merchant. This data is streamed from a source database and loaded into a ksqlDB table, keyed by the Geohash to a defined precision (the length of the hash). User location data is streamed from mobile devices and is joined to the merchant table by the Geohash. Location events that match are published to a "raw" alerts stream, which is further refined using the ksqlDB scalar function GEO_DISTANCE. This produces a final result of promo_alerts, which contains user and merchant data with geolocation information.
This tutorial assumes that you have merchant data stored in an SQL database. The merchant data includes geolocation information, which will be matched with the stream of location data from a user’s device. First, deploy a source connector that will read the merchant data into a Kafka topic for stream processing in ksqlDB.
|
|
When creating the initial STREAM or TABLE, if the backing Kafka topic already exists, then the PARTITIONS property may be omitted.
|
-- Substitute your parameter values in the connector configurations below.
-- If you do not want to connect to a real data source, remove the CREATE SOURCE CONNECTOR commands,
-- and add the INSERT INTO commands to insert mock data into the streams
CREATE SOURCE CONNECTOR IF NOT EXISTS recipe_postgres_merchant_data WITH (
'connector.class' = 'PostgresSource',
'kafka.api.key' = '<my-kafka-api-key>',
'kafka.api.secret' = '<my-kafka-api-secret>',
'connection.host' = '<database-host>',
'connection.port' = '5432',
'connection.user' = 'postgres',
'connection.password' = '<database-password>',
'db.name' = '<db-name>',
'table.whitelist' = 'merchant-locations',
'timestamp.column.name' = 'created_at',
'output.data.format' = 'JSON',
'db.timezone' = 'UTC',
'tasks.max' = '1'
);
SET 'auto.offset.reset' = 'earliest';
-- Creates a table of merchant data including the calculated geohash
CREATE TABLE merchant_locations (
id INT PRIMARY KEY,
description VARCHAR,
latitude DECIMAL(10,7),
longitude DECIMAL(10,7),
geohash VARCHAR
) WITH (
KAFKA_TOPIC = 'merchant-locations',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
);
-- Creates a table to lookup merchants based on a
-- substring (precision) of the geohash
CREATE TABLE merchants_by_geohash
WITH (
KAFKA_TOPIC = 'merchant-geohash',
FORMAT = 'JSON',
PARTITIONS = 6
) AS
SELECT
SUBSTRING(geohash, 1, 6) AS geohash,
COLLECT_LIST(id) as id_list
FROM merchant_locations
GROUP BY SUBSTRING(geohash, 1, 6)
EMIT CHANGES;
-- Creates a stream of user location data including the calculated geohash
CREATE STREAM user_locations (
id INT,
latitude DECIMAL(10,7),
longitude DECIMAL(10,7),
geohash VARCHAR
) WITH (
KAFKA_TOPIC = 'user-locations',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
);
-- Creates a stream of alerts when a user's geohash based location roughly
-- intersects a collection of merchants locations from the
-- merchants_by_geohash table.
CREATE STREAM alerts_raw
WITH (
KAFKA_TOPIC = 'alerts-raw',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
) AS
SELECT
user_locations.id as user_id,
user_locations.latitude AS user_latitude,
user_locations.longitude AS user_longitude,
SUBSTRING(user_locations.geohash, 1, 6) AS user_geohash,
EXPLODE(merchants_by_geohash.id_list) AS merchant_id
FROM user_locations
LEFT JOIN merchants_by_geohash ON SUBSTRING(user_locations.geohash, 1, 6) =
merchants_by_geohash.geohash
PARTITION BY null
EMIT CHANGES;
-- Creates a stream of promotion alerts to send a user when their location
-- intersects with a merchant within a specified distance (0.2 KM)
CREATE STREAM promo_alerts
WITH (
KAFKA_TOPIC = 'promo-alerts',
VALUE_FORMAT = 'JSON',
PARTITIONS = 6
) AS
SELECT
alerts_raw.user_id,
alerts_raw.user_geohash,
merchant_locations.description AS merchant_description,
CAST(
GEO_DISTANCE(alerts_raw.user_latitude, alerts_raw.user_longitude,
merchant_locations.latitude, merchant_locations.longitude,
'KM') * 1000 AS INT) as distance_meters,
STRUCT(lat := CAST(alerts_raw.user_latitude AS DOUBLE), lon := CAST(alerts_raw.user_longitude AS DOUBLE)) AS geopoint
FROM alerts_raw
LEFT JOIN merchant_locations on alerts_raw.merchant_id = merchant_locations.id
WHERE GEO_DISTANCE(
alerts_raw.user_latitude, alerts_raw.user_longitude,
merchant_locations.latitude, merchant_locations.longitude, 'KM') < 0.2
PARTITION BY null
EMIT CHANGES;
