NiFi PutFile установленное имя файла не работает

Question

1. Получение данных JSON через брокера Kafka.
2. Данные представлены в следующем формате, а данные изображения кодируются в Base64. например) {"filename":"test.jpg","filedata":"/9j/4AAQSkZJRgABAQEAYABgA....."}
3. Я хочу сохранить данные изображений, полученные через Kafka, в виде файла.
4. Однако он не работает должным образом.

Ниже приведен порядок, в котором я написал поток и описывает только ключевые настройки.

1. Процессор ConsumeKafka_2_0

2. EvaluateJsonPath Processor

   Flowfile-content назначения

   rawbytes $ .filedata

3. EvaluateJsonPath Processor (ошибка: не было допустимого содержимого JSON)

   атрибут потока-назначения

   имя файла $ .filename

4. Base64EncodeContent процессор

5. PutFile процессор

Когда поток выполняется, файл изображения сохраняется в обычном режиме, но имя файла установить нельзя. Что мне делать?

У вас есть сайт или примеры, на которые можно сослаться?

Сайт ссылки: https://community.hortonworks.com/articles/218015/ingesting-binary-files-like-pdf-jpg-png-to-hbase-w.html

Answer 1

На шаге 2 вы заменили содержимое потокового файла значением $ .filedata, которое больше не является JSON, поэтому вы не можете использовать EvaluateJsonPath на шаге 3, поскольку больше нет JSON.

Если вы измените шаги 2 и 3 в обратном порядке, вы можете извлечь имя файла в атрибут, сохранив исходный JSON в содержимом файла потока, а затем извлечь данные файла в содержимое.

Answer 2

Из вопроса я понял, что есть тема kafka, в которой есть имя файла и кодированное в base64 содержимое файла в формате json; Вы хотите использовать тему kafka, декодировать содержимое файла с помощью Base64, чтобы создать изображение, и сохранить изображение в имени файла, используя PutFile.

Я придумал поток, который выполнит это требование и говорит сам за себя.

1. ConsumeKafkaRecord_2_0 (использует {"имя файла": "test.jpg", "filedata": "/ 9j / 4AAQSkZ .."})
2. EvaluateJsonPath
   Назначение: атрибут flowfile
   rawtypes: $ .filedata
   имя файла: $ .filename
   
3. ReplaceText (Изменение содержимого потокового файла на содержимое кодированного изображения для следующего процессора)
4. Base64EncodeContent (rawtypes декодируется в изображение этим процессором)
5. UpdateAttribute (имя файла для хранения изображения обновляется здесь)
6. PutFile

Невозможно загрузить шаблон потока здесь. Размещение скриншота ключевых процессоров

EvaluateJSONPath

ReplaceText - (Заметьте значение замены)

UpdateAttribute

Answer 3

в соответствии с PutFile документацией:

Считывает атрибуты filename: имя файла, которое следует использовать при записи FlowFile на диск.

вам просто нужно использовать UpdateAttribute процессор, чтобы установить значение для filename атрибут