ksqlDB: SQL над потоками

В прошлом уроке мы посчитали заказы по зонам в минутных окнах на Kafka Streams. Это работает, но счётчик живёт в Java-приложении: его собирают, разворачивают и обновляют, как любой сервис. Отто же знает SQL ещё с курса аналитика и спрашивает, нельзя ли написать то же самое запросом. Можно, и сегодня разберём как.

Рецепт и повар

Можно готовить ужин самому: нарезать, обжарить, посолить, следить за плитой. А можно отдать повару рецепт: «овощи, обжарить 10 минут, посолить». Повар сделает то же самое, а ты описал только результат.

Kafka Streams это готовка самому: ты пишешь код, каждый шаг. ksqlDB это рецепт для повара: ты пишешь SQL, а сервер сам превращает его в такую же программу на Kafka Streams и запускает её.

Что такое ksqlDB

ksqlDB это отдельный сервер рядом с Kafka. Ему отправляют SQL, а он читает и пишет топики. Внутри каждого долгоживущего запроса работает приложение на Kafka Streams, то самое из прошлого урока. ksqlDB делает компания Confluent и выпускает под своей лицензией Confluent Community, а не под Apache 2.0, как саму Kafka.

В ksqlDB два вида объектов:

• Поток (STREAM): события, которые только добавляются. Каждый заказ это новая строка.
• Таблица (TABLE): текущее значение по каждому ключу. Новое значение заменяет старое, как в компактированном топике.

Поток заказов

Всё ниже мы выполнили на настоящем ksqlDB 0.29 поверх apache/kafka:4.3.1. Сначала объявим поток над топиком order-events:

CREATE STREAM order_events (zone VARCHAR KEY, order_id VARCHAR)
  WITH (KAFKA_TOPIC = 'order-events', VALUE_FORMAT = 'JSON', PARTITIONS = 1);

zone VARCHAR KEY говорит, что зона лежит в ключе сообщения, а order_id в значении, в JSON. Формат значения важен: мы пробовали сначала VALUE_FORMAT = 'KAFKA', то есть голую строку, и ksqlDB отказался группировать такой поток: This format does not yet support GROUP BY.

Положим в поток те же семь заказов, что в прошлом уроке. Время события задаёт служебная колонка ROWTIME, в миллисекундах:

INSERT INTO order_events (ROWTIME, zone, order_id) VALUES (1791720010000, 'центр', 'order-1');
INSERT INTO order_events (ROWTIME, zone, order_id) VALUES (1791720040000, 'центр', 'order-2');
-- и так далее: 12:00:50 уралмаш, 12:01:20 центр,
-- опоздавшие 12:00:55 и 12:00:58, между ними 12:02:10