NiFi技术问询：如何获取指定目录列表、排序并按目录读取文件？

完整处理流程 #

1. 提取符合日期格式的一级子目录 #

使用ListFile处理器，配置如下：

• Input Directory: /my_src
• Recurse Subdirectories: false（仅扫描一级子目录）
• Include Directories: true（让处理器列出目录而非文件）
• Filter: ^\d{8}$（正则匹配8位数字日期格式的目录，自动过滤不符合的目录）
• 执行后，每个符合条件的目录会生成一个FlowFile，包含核心属性：
  • absolute.path: 目录的完整路径（如/my_src/20211125）
  • filename: 目录名（如20211125）

2. 按日期对目录排序 #

使用Sort处理器，配置：

• Sort By: filename（基于目录名字符串排序，8位日期字符串的排序逻辑与日期新旧一致）
• Sort Order: Ascending（旧目录优先）或Descending（新目录优先）
• Sorting Strategy: 选择FlowFile Attribute，基于属性值完成排序

3. 动态读取每个目录下的文件 #

由于GetFile无法动态指定路径，改用ListFile + FetchFile组合实现：

• 在排序后的流程后添加ListFile处理器，将其Input Directory设置为表达式语言${absolute.path}，这样每个FlowFile（对应一个目录）会触发该处理器扫描对应目录下的所有文件
• 开启该ListFile的Recurse Subdirectories为false（仅扫描当前目录文件），可按需设置Filter过滤特定文件格式
• 接着用Sort处理器对文件排序：按filename属性选择升序/降序，实现文件名排序
• 最后用FetchFile处理器读取文件内容，File to Fetch设置为${absolute.path}/${filename}即可

替代方案：用ExecuteScript简化流程 #

如果觉得多处理器组合繁琐，可使用ExecuteScript（Groovy脚本）一次性完成目录遍历、排序和文件读取触发：

import org.apache.nifi.processor.io.StreamCallback
import java.nio.file.Files
import java.nio.file.Paths
import java.util.stream.Collectors

def flowFile = session.get()
if (!flowFile) return

def baseDir = "/my_src"
// 列出符合日期格式的目录并排序（默认升序，降序可替换为.sorted(Comparator.reverseOrder())）
def dirs = Files.list(Paths.get(baseDir))
                .filter { Files.isDirectory(it) }
                .filter { it.getFileName().toString() ==~ /\d{8}/ }
                .sorted()
                .collect(Collectors.toList())

dirs.each { dir ->
    def newFlowFile = session.create()
    newFlowFile = session.putAttribute(newFlowFile, "directory.path", dir.toString())
    session.transfer(newFlowFile, REL_SUCCESS)
}

session.remove(flowFile)

将该脚本配置到ExecuteScript后，会生成每个目录的FlowFile，后续可直接传递给ListFile（Input Directory设为${directory.path}）继续处理文件

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