Как выполнить агрегацию сообщений (событий) в Кафке?

Question

У меня есть следующие атрибуты:

1. OrgId
2. DeviceId
3. ResponseId

Я хочу агрегировать количество устройствдля организации и нет ответов для организации каждый раз, когда приходит ответ.Я не могу придумать подходящую структуру для агрегации.

Answer 1

Это довольно широкий вопрос.Оставив на время выбор технологии, если бы выразить ответ в псевдо-SQL, это выглядело бы так:

Я хочу объединить количество устройств для организации

SELECT OrgId,COUNT DISTINCT(DeviceId) FROM inbound_data GROUP BY OrgId

Количество ответов для организации каждый раз, когда приходит ответ

SELECT OrgId, COUNT DISTINCT(ResponseId) FROM inbound_data GROUP BY OrgId

Так что теперь это просто, как реализовать это.Если вы уже используете Apache Kafka, у вас есть API-интерфейс Kafka Streams для этого на Java, а также KSQL для этого на SQL.Я не уверен, что COUNT DISTINCT поддерживается пока.