CODEWITHSUNDEEP
apache-nifi
user1675314
user1675314

Reputation:

Nifi record the maximum timestamp

I have the timestamp value in each flow file, I need to compare the timestamp of the current flow files with previous flow files and if it is greater write that to a file using the Put file processor, at the end I need to have the max timestamp in the file. I tried this with the update attribute processor's advanced features in the following way:

Added a rule named max_timestamp with the condition ${getStateValue("maxTimestamp"):lt(${timestamp})} and set the action for the attribute maxTimestamp as ${timestamp}

I also turned set the store state feature to store locally in the processor, but I do not see the maxTimestamp attribute being set to the flowfiles output from the update attribute processor.

Upvotes: 0

Views: 2273

Answers (1)

daggett
daggett

Reputation: 28564

  1. use UpdateAttribute processor to evaluate maxTimestamp attribute in stateful mode with following parameters:
Store State                      : Store state locally
Stateful Variables Initial Value : 0
maxTimestamp                     : ${timestamp:toNumber():math("max",${getStateValue("maxTimestamp")})}

the expression ${timestamp:toNumber():math("max",${getStateValue("maxTimestamp")})}

evaluates maximum between two numbers:

timestamp attribute in current flowFile, and stored value of maxTimestamp

The property Stateful Variables Initial Value : 0 defines the default value of the stored value maxTimestamp

  1. then use RouteOnAttribute with following properties:
Routing Strategy : Route to Property name
less_then        : ${timestamp:lt(${maxTimestamp})}

So, you will have 2 relationships

less_then for case when timestamp is less then maxTimestamp

unmatched for other cases

Upvotes: 1

Related Questions