ksqlDB: база данных, созданная специально для приложений потоковой обработки

Обзор

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

• Streams и tables — создание отношений со схемами для данных вашей темы Apache Kafka;
• Материализованные представления — определение материализованных представлений в реальном времени, которые обновляются инкрементально, над потоками с использованием SQL;
• Запросы push — непрерывные запросы, которые отправляют инкрементные результаты клиентам в режиме реального времени;
• Запросы pull — запрос материализованных представлений по требованию, как в традиционной базе данных;
• Connect — интеграция с любым источником или приёмником данных Kafka Connect полностью из ksqlDB.

Комбинация этих мощных примитивов позволяет создавать полноценное потоковое приложение только с помощью операторов SQL, минимизируя сложность и операционные издержки. ksqlDB поддерживает широкий спектр операций, включая агрегацию, объединение, оконные функции, сегментацию сеансов и многое другое. Вы можете найти дополнительные руководства и ресурсы ksqlDB здесь.

Начало работы

• Следуйте руководству по быстрому запуску ksqlDB, чтобы начать работу за несколько минут.
• Ознакомьтесь с документацией ksqlDB.
• Посмотрите на некоторые рецепты использования ksqlDB для примеров распространённых шаблонов.

Документация

См. документацию ksqlDB для последней стабильной версии.

Примеры использования и примеры

Материализованные представления

ksqlDB позволяет определять материализованные представления над вашими потоками и таблицами. Материализованные представления определяются так называемым «постоянным запросом». Эти запросы известны как постоянные, потому что они поддерживают свои инкрементно обновляемые результаты с помощью таблицы.

CREATE TABLE hourly_metrics AS
  SELECT url, COUNT(*)
  FROM page_views
  WINDOW TUMBLING (SIZE 1 HOUR)
  GROUP BY url EMIT CHANGES;

Результаты можно «вытягивать» из материализованных представлений по запросу через SELECT запросы. Следующий запрос вернёт одну строку:

SELECT * FROM hourly_metrics
  WHERE url = 'http://myurl.com' AND WINDOWSTART = '2019-11-20T19:00';

Результаты также могут непрерывно «проталкиваться» клиентам через потоковые SELECT запросы. Следующий потоковый запрос будет отправлять клиенту все инкрементные изменения, внесённые в материализованное представление:

SELECT * FROM hourly_metrics EMIT CHANGES;

Потоковые запросы будут выполняться постоянно, пока они не будут явно остановлены.

Потоковая ETL

Apache Kafka является популярным выбором для питания конвейеров данных. ksqlDB упрощает преобразование данных внутри конвейера, подготавливая сообщения к чистому попаданию в другую систему.

CREATE STREAM vip_actions AS
  SELECT userid, page, action
  FROM clickstream c
  LEFT JOIN users u ON c.userid = u.user_id
  WHERE u.level = 'Platinum' EMIT CHANGES;

Обнаружение аномалий

ksqlDB хорошо подходит для выявления закономерностей или аномалий в данных в реальном времени. Обрабатывая поток по мере поступления данных, вы можете идентифицировать и правильно отображать необычные события с миллисекундной задержкой.

CREATE TABLE possible_fraud AS
  SELECT card_number, count(*)
  FROM authorization_attempts
  WINDOW TUMBLING (SIZE 5 SECONDS)
  GROUP BY card_number
  HAVING count(*) > 3 EMIT CHANGES;

Мониторинг

Способность Kafka обеспечивать масштабируемость... Организованный мониторинг и оповещение с помощью упорядоченных записей и потоковой обработки данных

ksqlDB предоставляет знакомый синтаксис для отслеживания, понимания и управления оповещениями.

CREATE TABLE error_counts AS
  SELECT error_code, count(*)
  FROM monitoring_stream
  WINDOW TUMBLING (SIZE 1 MINUTE)
  WHERE  type = 'ERROR'
  GROUP BY error_code EMIT CHANGES;

Интеграция с внешними источниками и приёмниками данных

ksqldb включает встроенную интеграцию с источниками и приёмниками данных Kafka Connect, обеспечивая унифицированный SQL-интерфейс для широкого спектра внешних систем (https://www.confluent.io/hub).

Следующий запрос представляет собой простой постоянный потоковый запрос, который будет выводить все свои результаты в тему с именем clicks_transformed:

CREATE STREAM clicks_transformed AS
  SELECT userid, page, action
  FROM clickstream c
  LEFT JOIN users u ON c.userid = u.user_id EMIT CHANGES;

Вместо того чтобы просто отправлять весь непрерывный вывод запроса в Kafka-тему, часто бывает очень полезно направить вывод в другое хранилище данных. Интеграция ksqlDB с Kafka Connect делает этот шаблон очень простым.

Следующее утверждение создаст коннектор приёмника Kafka Connect, который непрерывно отправляет весь вывод из вышеупомянутого потокового запроса ETL непосредственно в Elasticsearch:

 CREATE SINK CONNECTOR es_sink WITH (
  'connector.class' = 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector',
  'key.converter'   = 'org.apache.kafka.connect.storage.StringConverter',
  'topics'          = 'clicks_transformed',
  'key.ignore'      = 'true',
  'schema.ignore'   = 'true',
  'type.name'       = '',
  'connection.url'  = 'http://elasticsearch:9200');

Присоединяйтесь к сообществу

Для получения помощи, вопросов или запросов о ksqlDB используйте нашу пользовательскую группу Google (https://groups.google.com/forum/#!forum/ksql-users) или наш общедоступный канал Slack #ksqldb в Confluent Community Slack. Добро пожаловать всем!

Вы можете получить помощь, узнать, как внести свой вклад в ksqlDB, и найти последние новости, связавшись с сообществом Confluent.

По более общим вопросам о платформе Confluent обращайтесь в группу Google Confluent (https://groups.google.com/forum/#!forum/confluent-platform).

Вклад и сборка из исходного кода

Вклад в код, примеры, документацию и т. д. очень ценятся.

• Сообщайте о проблемах и ошибках непосредственно в этом проекте GitHub (https://github.com/confluentinc/ksql/issues).
• Узнайте, как работать с исходным кодом ksqlDB, включая сборку и тестирование ksqlDB, а также внесение изменений в код ksqlDB, прочитав наши рекомендации по разработке и вкладу (CONTRIBUTING.md).
• Один из хороших способов начать — заняться новой проблемой ([https://github.com/confluentinc/ksql/labels/good%20first%20issue](https://github.com/confluentinc/ksql/labels/good first issue)).

Лицензия

Проект лицензирован в соответствии с лицензией сообщества Confluent.

Apache, Apache Kafka, Kafka и связанные с ними названия проектов с открытым исходным кодом являются товарными знаками Apache Software Foundation.