merging of flow files in the specified order

Viewed 911

I am new to nifi(using version 1.8.0). I have the requirement of consuming kafka messages which contain vehicle position in the form of lat,lon per message. Since each message will arrive as a flow file, I need to merge all these flow files and make a json file containing the complete path followed by the vehicle. I am using consume kafka processor to subscribe to messages, update attribute processor(properties added are filename:${getStateValue("seq")},seq:${getStateValue("seq"):plus(1)}) to add a sequence number as filename (eg. filename is 1,2, 3 etc) and put file processor to write these files in the specified directory. I have configured FIFO priority queue on all the success relationship between the above mentioned processors.Once, I have received all the messages I want to merge all the flow files. For this I know I have to use get file, enforce order, merge content(merge strategy:bin packing algorithm, merge format:binary concatenation) and put file processor, respectively. Is my approach correct? How should I establish that merging of files takes place in the sequence of their names as filename is a seq number. What should I put in order attribute in enforce order processor?What should in put in group identifier? Are there more custom fields to be added in enforce order processor?

1 Answers

EnforceOrder processor documentation

1.Group Identifier

  • This property evaluate on each flowfile for your case use UpdateAttribute Processor, add group_name attribute and use the same ${group_name} attribute in Group Identifier property value.

2.Order Attribute

  • Expression language is not supported.

  • You can use filename (or) create new attribute in UpdateAttribute processor and use same attribute name in your Order Attribute property value.

For reference/usage of Enforce order processor use this template and upload to your NiFi instance.

Related