NiFi技术问询：如何通过触发文件触发ListFile批量移动有序文件

核心思路 #

通过WaitForFile监听触发文件，生成可传递上下文的FlowFile，再借助脚本+系统命令实现批量文件移动，完美适配“无法预知最终文件名/到达时间、需触发启动”的需求。

步骤实现 #

1. 监听触发文件（WaitForFile） #

配置WaitForFile处理器：

• Directory：填写存放part-xxx.csv和trigger.txt的源文件夹路径
• File Filter：设为trigger.txt
• Success Criteria：选择File Exists，确保触发文件一出现就启动流程
• 按需调整轮询间隔等参数

该处理器会在检测到trigger.txt后，生成携带源文件夹路径属性的FlowFile，作为后续流程的触发信号。

2. 获取待移动文件列表（ExecuteScript） #

用Groovy脚本读取源文件夹内所有符合命名规则的CSV文件，写入FlowFile并标记属性：

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

flowFile = session.write(flowFile, { outputStream ->
    def sourceDir = flowFile.getAttribute('absolute.path')
    // 筛选part-xxx.csv格式的文件
    def filePaths = Files.list(Paths.get(sourceDir))
        .filter(p -> p.getFileName().toString() ==~ /part-\d+\.csv/)
        .map(p -> p.toString())
        .collect(Collectors.joining('\n'))
    outputStream.write(filePaths.getBytes('UTF-8'))
} as StreamCallback)

// 添加文件列表属性，供后续命令调用
flowFile = session.putAttribute(flowFile, 'file.list', flowFile.getAttribute('content'))
session.transfer(flowFile, REL_SUCCESS)

3. 批量移动文件（ExecuteProcess） #

调用Windows批处理命令完成批量移动：

• Command：cmd.exe
• Arguments：/c "for %f in (${file.list}) do move \"%f\" \"你的目标文件夹绝对路径\""

注：目标路径若含空格，需用双引号包裹；${file.list}会自动替换为前一步生成的文件路径列表。

4. 清理触发文件（可选） #

添加DeleteFile处理器，配置路径为trigger.txt的绝对路径，或从FlowFile属性中读取路径，完成移动后自动删除触发文件。

流程链路 #

WaitForFile → ExecuteScript → ExecuteProcess →（可选）DeleteFile

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