nifi как производитель для kafka: данные не являются последовательными при чтении Kafka

Я публикую потоковые файлы из nifi в kafka, используя процессор publishKafka_0_10. При чтении данных из Kafka через код последовательность данных не сохраняется (сортируется по отметке времени). Мой набор данных выглядит так: метка времени, канал, значение.

Просто для отладки я публикую те же потоковые файлы в phoenix, используя PutSql, и вижу, что в таблице Phoenix данные последовательны (отсортированы по времени). Было бы здорово, если бы кто-нибудь объяснил мне, почему я не могу последовательно читать данные из kafka. В кафке в топике только один раздел. Заранее спасибо.


apache-kafka apache-nifi kafka-producer-api hortonworks-dataflow
person Swati    schedule 09.01.2018    source источник

Ответы (1)


arrow_upward
0
arrow_downward

Kafka гарантирует порядок только внутри раздела. Раз вы говорите, что это один раздел, то ладно.

Мой набор данных выглядит так: метка времени, канал, значение.

Временные метки сообщений — это просто записи метаданных (ваши собственные временные метки не передаются в класс Kafka ProducerRecord от NiFi). Кроме того, временные метки не влияют на порядок. Другими словами, если одно сообщение с «поздней временной меткой» фиксируется раньше другого «более раннего» времени, тогда да, оно хронологически не в порядке, но Кафка просто видит, что смещения сдвинулись.

почему я не могу последовательно читать данные из kafka

Да, но в том порядке, в котором сообщения были переданы Кафке.

Ваш потребительский код должен извлекать временные метки записи и соответствующим образом переупорядочивать их. Например, в Kafka Connect есть экстрактор Record Timestamp, который может записывать данные в секционированные каталоги на основе этого времени. Я предполагаю, что ваш процессор PutSQL читает последовательно упорядоченные файлы FlowFiles (которые имеют свои собственные временные метки, а не временные метки в ваших данных, если только вы не запускали процессор ModifyAttribute), не используя процессор ConsumeKafka?

person OneCricketeer    schedule 09.01.2018
comment
Я согласен. Тогда есть ли какой-либо процессор в nifi, с помощью которого я последовательно изменяю последовательность данных на основе моей временной метки Daa при создании потоковых файлов в Kafka. - person Swati; 09.01.2018
comment
Я быстро просмотрел исходный код процессора и думаю, что все PublishKafka под 0_10 используют один и тот же класс для создания, а временная метка, насколько я могу судить, не поддается изменению. - person OneCricketeer; 09.01.2018
comment
Это означает, что единственное решение состоит в том, что при потреблении сообщений читателю нужно переставлять их по порядку? - person Swati; 09.01.2018
comment
При использовании NiFi это выглядело бы так. Как я уже упоминал, Kafka Connect — это фреймворк, который учитывает время сообщений Kafka. К сожалению, нет интерфейса, подобного NiFi. - person OneCricketeer; 09.01.2018
comment
NiFi собирается публиковать данные в Kafka, читая потоковый файл построчно или запись за записью и публикуя каждую из них, однако данные организованы в потоковом файле. - person Bryan Bende; 09.01.2018
comment
@cricket_007 cricket_007, если бы NiFi мог установить поле метки времени в каждой записи ProducerRecord, вы говорите, что Кафка уважал бы это и переупорядочивал данные на стороне брокера? Это будет работать только для каждого коммита, верно? потому что журнал предназначен только для добавления, поэтому он может только переупорядочивать пакет записей, которые должны быть зафиксированы. - person Bryan Bende; 09.01.2018
comment
@Byran Я думаю, что ОП спрашивает о порядке времени в самих данных. Есть также время FlowFile и время Kafka Record. Глядя на исходный код, я могу сказать, что время записи Kafka напрямую не связано со временем генерации FlowFile или данными, а скорее со временем, когда запись Kafka создается во время передачи. Все, что я говорю, это то, что если бы NiFi мог записать временную метку, можно было бы сопоставить время данных или FlowFile, в зависимости от параметра конфигурации. Постараюсь не предъявлять существенных претензий к заказу, так как не пробовал - person OneCricketeer; 09.01.2018