如何在nifi的putfile处理器上维护唯一的时间戳

rfbsl7qr  于 2021-07-12  发布在  Spark
关注(0)|答案(1)|浏览(576)

我有一些csv数据需要放在一个位置与特定的文件命名约定唯一的文件名。
流动顺序
合并记录->在这里csv行将被合并并前进到更新属性处理器。
update attribute->当合并的flowfile内容(csv行的集合)流经update attribute处理器时,具有以下语法的当前时间戳将被分配给“filename”flowfile属性。语法:test-${now():format(“yyyymmddhhmmssss”,“ist”)}.csv
putsftp->现在在putsftp服务器上,无论从update属性接收的流文件发布到远程服务器。
发行声明:
我的远程服务器对文件名格式有限制,它应该是特定的,定义如下:语法:test-${now():format(“yyyymmddhhmmssss”,“ist”)}.csv egg:test-202104041836555.csv
因此,在更新属性处理器上,由于线程在同一时间戳示例中处理多个流文件,因此多个流文件(合并内容)被分配了相同的文件名,并且在放入sftp服务器时,csv文件无法处理并放入远程服务器,因为已经存在同名文件。
注意:我与putsftp处理器中的冲突解决策略没有任何关系,因为我的sftp服务器客户端对文件命名格式有很强的依赖性。
处理器序列流的参考图像:

更新属性处理器属性:

nnt7mjpx

nnt7mjpx1#

我通过引入控制处理器找到了解决方案。controlrate处理器配置:

通过在一个示例/时间点仅向udpate属性处理器泵送一个流文件,am能够实现为每个流文件分配唯一的时间戳。
更新流程图:

相关问题