Доступ к содержимому FlowFile в процессоре NIFI PutS3Object

Question

Я новичок в NIFI и хочу передавать данные из Kafka в ведро S3. Я использую процессор PutS3Object и могу передавать данные на S3, если я жестко кодирую значение Bucket как mphdf/orderEvent, но я хочу указать ведра на основе поля в содержимом FlowFile, который находится в Json. Итак, если содержание Json это {"menu": {"type": "file","value": "File"}}, могу ли я иметь значение для свойства Bucket как mphdf/$.menu.type? Я попытался сделать это и получить ошибку ниже. Я хочу знать, есть ли способ доступа к содержимому FlowFile с процессором PutS3Object и сделать имена Bucket настраиваемыми или мне нужно будет создать собственный процессор?

ERROR [Timer-Driven Process Thread-10] 
o.a.nifi.processors.aws.s3.PutS3Object 
com.amazonaws.services.s3.model.AmazonS3Exception: The XML you 
provided was not well-formed or did not validate against our 
published schema (Service: Amazon S3; Status Code: 400; Error Code: 
MalformedXML; Request ID: 77DF07828CBA0E5F) 

Answer 1

Я считаю, что вы хотите сделать, это использовать EvaluateJSONPath процессор, который оценивает произвольные выражения JSONPath против содержания JSON и извлекает результаты flowfile атрибуты. Затем вы можете ссылаться на атрибут flowfile с использованием языка выражения NiFi в конфигурации PutS3Object (см. Ваше первое свойство Ключ объекта, который ссылается на ${filename}). Таким образом, вы бы оценить $.menu.type и хранить его в атрибут menuType в EvaluateJSONPath процессоре, то в PutS3Object вы бы Ковш быть mphdf/${menuType}.

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