Dealing with Kafka

1. Create inbound Kafka topic

kafka-topics \\
--bootstrap-server kafka-headless:9092 \\
--topic readings \\
--create --partitions 6 \\
--replication-factor 3

1. Create target table
2. Create Kafka Engine table

CREATE TABLE readings_queue (
readings_id Int32,
time DateTime,
temperature Decimal(5,2)
) ENGINE = Kafka SETTINGS
 kafka_broker_list = 'kafka-headless.kafka:9092',
 kafka_topic_list = 'readings',
 kafka_group_name = 'readings_consumer_group1',
 kafka_num_consumers = '1',
 kafka_format = 'CSV'

1. Create materialized view

Arrays

metrics_value Array(Float64),

Array are like in Python except the index of array:

WITH [1, 2, 4] AS array
SELECT
 array[1] AS First,
 array[2] AS Second,
 array[3] AS Third,
 array[-1] AS Last,
 length(array) AS Length

ARRAY JOIN unrolls array to rows

Useful functions:

• groupArray — combine values from rows to one array
• arraySort — sort array in ascending order (also can be used with lambda functions)

Processing JSON data