У меня есть этот поток NiFi, который захватывает события в JSON от брокера MQTT, группирует их в соответствии с некоторыми критериями, преобразует их в строки Avro и должен выводить их через файлы в кластере Had oop.
Я выбрал Avro в качестве формата хранения, поскольку он может добавлять новые данные в существующий файл.
Эти события сгруппированы по источникам, и в идеале у меня должен быть один отдельный файл Avro в HDFS для каждого источника событий Таким образом, NiFi накапливает новые события в каждом файле по мере их появления (конечно, с правильной пакетной записью, поскольку выдача записи для каждого нового события не будет очень хорошей, я уже работал с этим с процессором MergeContent).
У меня есть отработанный поток, но я обнаружил, что последний шаг, процессор PutHDFS, - это формат файла c, то есть он не понимает, как добавить файл в существующий файл Avro.
Я нашел этот запрос на получение , который реализует именно это, но он никогда не был объединен с NiFi из-за различных проблем.
* 10 12 * Есть ли способ сделать это с существующими процессорами NiFi? Или мне нужно развернуть свой собственный процессор PutHDFS, который понимает, как добавить к существующим файлам Avro?