Стримы Kafka - первый пример WordCount неправильно считает первый круг

Я изучаю Kafka Streams, и у меня проблема с первым примером WordCount в Java 8, взятым из документации.

Использование последних доступных версий потоков kafka, пример лямбда-выражений Kafka Connect и WordCount.

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

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

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

Как случилось, что первый раз посчитали неправильно?

Пример [ПЕРВЫЕ ДАННЫЕ]: (введите тему в Kafka) привет привет микрофон микрофон тест

(Стриминг приложения запущен)

(тема вывода) привет 12 микрофон 4 тест 3 (случайные подсчеты)

[ПОСЛЕДНИЕ ДАННЫЕ - Публикация во входной теме одинаковых слов]

(выходная тема) привет 14 майк 6 тест 4

[НОВАЯ ПОПЫТКА]

(выходная тема) привет 16 майк 8 тест 5

и так далее....


apache-kafka apache-kafka-connect bigdata apache-kafka-streams
person Aniello Guarino    schedule 08.03.2017    source источник
comment
Звучит странно. Можете ли вы достоверно воспроизвести проблему? Этого не должно происходить.   -  person Matthias J. Sax    schedule 08.03.2017

Ответы (1)


arrow_upward
3
arrow_downward

Демонстрация WordCount в Apache Kafka имеет следующие строки:

// setting offset reset to earliest so that we can re-run the demo code with the same pre-loaded data
// Note: To re-run the demo, you need to use the offset reset tool:
// https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Это означает, что при перезапуске приложения оно будет считывать тему ввода с самого начала («самого раннего») iff, для приложения WordCount, хранящегося в Kafka, нет существующих потребительских смещений. Смещения потребителя приложения истекают в Kafka после определенного периода бездействия приложения, по умолчанию - 24 часа (см. offsets.retention.minutes конфигурация брокера).

Я мог представить, что произошло следующее:

  • Некоторое время назад вы экспериментировали с Kafka и вводили тестовые данные во входную тему.
  • Затем вы сделали перерыв> 24 часов перед возобновлением экспериментов.
  • Теперь приложение при перезапуске вернулось к повторному чтению темы ввода с самого начала, тем самым подбирая старые входные данные теста и, таким образом, приводя к «завышенным» счетчикам.

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

Вы можете проверить мою гипотезу выше, снова посмотрев на тему ввода с помощью потребителя консоли, добавив параметр CLI --from-beginning (см. https://kafka.apache.org/documentation/#quickstart_consume).

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic yourInputTopic --from-beginning

Это покажет вам все доступные данные в теме "yourInputTopic" - за вычетом любых данных, которые могли быть удалены из тем Kafka за это время (конфигурация брокера по умолчанию будет очищать данные старше 7 дней, см. log.retention.hours) .

person Michael G. Noll    schedule 09.03.2017
comment
Спасибо за ваш ответ. На самом деле, когда я тестировал через 24 часа (затем новые смещения), я удалял старую тему (я включил отмену) и воссоздавал их с нуля для нового чистого выполнения. Проблема возникла снова. Но теперь я добавил строку streamsConfiguration.put (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, самое раннее) в пример кода (не было раньше) и, похоже, работает хорошо. Может я не решил именно с этим, но работает. - person Aniello Guarino; 09.03.2017
comment
У меня была аналогичная проблема несколько недель назад, но иногда счет был отрицательным. Может ли это быть вызвано чем-то подобным? - person foxygen; 17.03.2017
comment
Отрицательные счета? Похоже на другую проблему. Проблема, описанная здесь, должна приводить только к завышению количества (т. Е. К количеству, которое больше, чем должно быть на самом деле). - person Michael G. Noll; 18.03.2017