Nifi记录最大时间戳

Question

我在每个流文件中都有时间戳值，我需要将当前流文件的时间戳与以前的流文件进行比较，如果更大，则使用 Put 文件处理器将其写入文件，最后我需要在文件中有最大时间戳。我通过以下方式使用更新属性处理器的高级功能进行了尝试：

添加了一个名为 max_timestamp 的规则，条件 ${getStateValue("maxTimestamp"):lt(${timestamp})} 并为属性 maxTimestamp as ${timestamp}

设置了操作

我也将存储状态功能设置为本地存储在处理器中，但我没有看到 maxTimestamp 属性被设置为更新属性处理器输出的流文件。

Answer 1

使用 UpdateAttribute 处理器在有状态模式下使用以下参数评估 maxTimestamp 属性：

Store State                      : Store state locally
Stateful Variables Initial Value : 0
maxTimestamp                     : ${timestamp:toNumber():math("max",${getStateValue("maxTimestamp")})}

表达式${timestamp:toNumber():math("max",${getStateValue("maxTimestamp")})}

计算两个数字之间的最大值：

timestamp 当前流文件中的属性，和 maxTimestamp

的存储值

属性Stateful Variables Initial Value : 0定义存储值的默认值maxTimestamp

然后使用具有以下属性的 RouteOnAttribute：

Routing Strategy : Route to Property name
less_then        : ${timestamp:lt(${maxTimestamp})}

所以，你将有 2 个关系

less_then 对于 timestamp 小于 maxTimestamp

的情况

unmatched其他情况

Nifi记录最大时间戳

Nifi record the maximum timestamp

apache-nifi