Hazelcast SQL Kafka + IMap JOIN

Örnek

Kafka için mapping yaratırız.

CREATE OR REPLACE MAPPING pizzastream(
  timestamp_ TIMESTAMP,
  pizza VARCHAR,
  user_id VARCHAR,
  quantity DOUBLE
)
TYPE Kafka
OPTIONS (
  'keyFormat' = 'varchar',
  'valueFormat' = 'json-flat',
  'auto.offset.reset' = 'earliest',
  'bootstrap.servers' = 'localhost:19092'
);

IMap için mapping yaratırız.

CREATE or REPLACE MAPPING recommender (
  __key BIGINT,
  user_id VARCHAR,
  extra1 VARCHAR,
  extra2 VARCHAR,
  extra3 VARCHAR )
TYPE IMap
OPTIONS (
  'keyFormat'='bigint',
  'valueFormat'='json-flat'
);

IMap'e veri yazarız

INSERT INTO recommender VALUES
(1, 'user_1', 'Soup','Onion_rings','Coleslaw'),
(2, 'user_2', 'Salad', 'Coleslaw', 'Soup'),
(3, 'user_3', 'Zucchini_fries','Salad', 'Coleslaw'),
(4, 'user_4', 'Onion_rings','Soup', 'Jalapeno_poppers'),
(5, 'user_5', 'Zucchini_fries', 'Salad', 'Coleslaw'),
(6, 'user_6', 'Soup', 'Zucchini_fries', 'Coleslaw'),
(7, 'user_7', 'Onion_rings', 'Soup', 'Jalapeno_poppers'),
(8, 'user_8', 'Jalapeno_poppers', 'Coleslaw', 'Zucchini_fries'),
(9, 'user_9', 'Onion_rings','Jalapeno_poppers','Soup');

Join işlemi yaparız

SELECT
    pizzastream.user_id AS user_id,
    recommender.extra1 as extra1,
    recommender.extra2 as extra2,
    recommender.extra3 as extra3,
     pizzastream.pizza AS pizza
FROM pizzastream
JOIN recommender
ON recommender.user_id = recommender.user_id 
AND recommender.extra2 = 'Soup';

Join sonucunu tekrar Kafka'ya göndeririz.

CREATE OR REPLACE MAPPING recommender_pizzastream(
  timestamp_ TIMESTAMP,
  user_id VARCHAR,
  extra1 VARCHAR,
  extra2 VARCHAR,
  extra3 VARCHAR,
  pizza VARCHAR
)
TYPE Kafka
OPTIONS (
  'keyFormat' = 'int',
  'valueFormat' = 'json-flat',
  'auto.offset.rest' = 'earliest',
  'bootstrap.servers' = 'localhost:19092'
);

CREATE JOB recommender_job AS SINK INTO recommender_pizzastream SELECT
    pizzastream.timestamp_ as timestamp_,
    pizzastream.user_id AS user_id,
    recommender.extra1 as extra1,
    recommender.extra2 as extra2,
    recommender.extra3 as extra3,
    pizzastream.pizza AS pizza
FROM pizzastream
JOIN recommender
ON recommender.user_id = recommender.user_id 
AND recommender.extra2 = 'Soup';