如何在Apache NiFi中按文件名排序文件？处理器使用遇阻求助

针对你用GetFile拉取.json文件后，无法通过EnforceOrder实现多种排序的问题，核心原因是EnforceOrder仅支持数值型属性作为排序依据，字符串或未转换的日期属性无法生效。以下是四种排序场景的具体实现步骤：

一、按文件名排序 #

实现步骤： #

1. 在GetFile后添加ExecuteScript处理器，用Groovy脚本为每个文件生成按字典序排序的序号：

   def flowFiles = session.get(100)
   if (!flowFiles) return
   
   // 收集所有文件并按文件名字典序排序
   def sortedFiles = flowFiles.collect { ff ->
       [flowFile: ff, filename: ff.getAttribute('filename')]
   }.sort { a, b -> a.filename <=> b.filename }
   
   // 为排序后的文件分配递增序号属性
   def order = 0
   sortedFiles.each { entry ->
       entry.flowFile.setAttribute('filename_order', String.valueOf(order++))
       session.transfer(entry.flowFile, REL_SUCCESS)
   }
   

2. 添加EnforceOrder处理器，配置：
   • Order Attribute：filename_order
   • Start Order Value：0
   • Allow Reordering：true
   • Maximum Unordered Count：设置为待排序文件总数（或足够大的数值）

二、按文件名中的日期转换后排序 #

假设文件名格式为data_20240520.json（含yyyyMMdd格式日期）：

实现步骤： #

1. 添加ExtractText处理器，提取日期字符串：
   • Regular Expression：data_(\d{8})\.json
   • Capture Group 1：file_date_str
2. 添加UpdateAttribute处理器，将日期字符串转为毫秒时间戳（数值型）：
   • 新增属性file_date_timestamp，值为：

     ${file_date_str:toDate('yyyyMMdd'):toNumber()}
     

3. 配置EnforceOrder：
   • Order Attribute：file_date_timestamp
   • 其余参数参考场景一设置

三、按文件实际修改日期排序 #

之前失败是因为file.lastModifiedTime是字符串格式，需转为数值型时间戳：

实现步骤： #

1. 添加UpdateAttribute处理器，生成数值型修改时间戳：
   • 新增属性file_modified_timestamp，值为：

     ${file.lastModifiedTime:toDate():toNumber()}
     

2. 配置EnforceOrder：
   • Order Attribute：file_modified_timestamp
   • 其余参数参考场景一设置

四、按正则分组后排序（先分组，组内排序） #

假设文件名格式为groupA_20240520.json，需按groupA/groupB分组后，组内按日期排序：

实现步骤： #

1. 添加ExtractText处理器，提取分组标识：
   • Regular Expression：(group[A-Z])_.*\.json
   • Capture Group 1：file_group
2. 添加RouteOnAttribute处理器，按file_group路由到不同队列：
   • 为每个分组（如groupA）添加路由规则：${file_group:equals('groupA')}，对应独立的关系队列
3. 对每个分组队列，重复场景二的步骤（提取日期转时间戳+EnforceOrder排序）
4. 如需合并分组结果，可使用MergeContent处理器汇总各分组的排序后文件

关键配置注意事项 #

• Allow Reordering必须设为true：流文件到达顺序通常与排序顺序不一致，开启后处理器会缓存未按序到达的文件，直到顺序匹配
• Maximum Unordered Count需足够大：建议设置为待排序文件总数，避免因缓存不足导致文件丢失
• 所有排序属性必须是数值型：EnforceOrder不支持字符串排序，务必将所有排序依据转为整型/长整型属性

内容的提问来源于stack exchange，提问作者edjm