Вопрос в руководстве разработчика для Kafka dsl api
У меня возник вопрос, когда я шел по этому замечательному руководству https://kafka.apache.org/20/documentation/streams/developer-guide/dsl-api.html
Мой вопрос находится в разделе «Пример семантики для агрегации таблиц». В частности, посмотрите на таблицу в этом разделе, на отметке времени 4, но каков механизм выполнения агрегатором "(E, 5 - 5)".
Меня смущает то, что ключ уже преобразован из имени («Алиса») в регион («А») на этапе группировки. Как «groupedTable» все еще может воспринимать исходный ключ в совокупности и выполнять вычитание?
Заранее спасибо.
Ответы 1
Здесь действуют два механизма:
- базовое хранилище может получить старое значение ключа из хранилища, прежде чем оно поместит новое значение в хранилище
- при необходимости вышестоящий оператор, на котором размещено базовое хранилище, отправит нижестоящему оператору как новое, так и старое значение.
Не уверен, что смогу следовать. Однако отправка старого и нового значения является внутренней деталью реализации и не отображается в API (если это был ваш вопрос). Дело в том, что внутри поток данных между userProfiles KTable и aggregated KTable не является простым KeyValue.pair(region, user.length()), как указано в groupBy(), фактически физически передаваемые данные представляют собой две вложенные пары <key,<newValue,oldValue>>.
Спасибо, Матиас, что ответил на мой вопрос! Хотя мне потребуется некоторое время, чтобы переварить его значение. Любая ссылка на эту тему приветствуется.
Это не задокументировано, потому что это внутренние детали реализации - вы можете только заглянуть в код...
Спасибо! Но я не могу соотнести эти механизмы с кодом, представленным в руководстве (прямо над таблицей). Я предполагаю, что содержимое таблицы должно отражать логику кода.