如何在NiFi完成业务逻辑后将源文件从in_progress移至completed？

首先明确：步骤7需要移动的是本地文件系统中in_progress目录下的原物理文件，而非NiFi的FlowFile内容，核心是让FlowFile携带该原文件的准确路径，以便业务逻辑完成后能定位并移动它。

前置准备（确保FlowFile携带原文件路径） #

在步骤5将文件移至in_progress目录时，必须把该文件的完整绝对路径存入FlowFile的自定义属性中（比如命名为in_progress_file_path）：

• 若用ExecuteProcess执行系统移动命令（如Linux的mv），执行后通过UpdateAttribute处理器，将目标路径拼接为属性值（例如把${pending_dir}/${filename}替换为${in_progress_dir}/${filename}）。
• 若用ExecuteScript（Groovy）执行移动，脚本中直接将新路径写入FlowFile属性。

步骤7的具体实现方案 #

方案1：用ExecuteScript处理器（Groovy脚本推荐） #

灵活性强，可处理异常场景：

1. 添加ExecuteScript处理器，选择Groovy语言。
2. 写入以下脚本（根据实际路径调整）：

import java.nio.file.Files
import java.nio.file.Paths
import java.nio.file.StandardCopyOption

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

def inProgressPath = flowFile.getAttribute('in_progress_file_path')
def completedDir = '/path/to/completed' // 可改为从NiFi参数或属性读取
def completedPath = Paths.get(completedDir, flowFile.getAttribute('filename'))

try {
    Files.move(Paths.get(inProgressPath), completedPath, StandardCopyOption.REPLACE_EXISTING)
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    log.error("移动文件失败: ${e.getMessage()}", e)
    session.transfer(flowFile, REL_FAILURE)
}

3. 配置处理器关系：将REL_SUCCESS路由到正常后续流程，REL_FAILURE路由到错误处理分支（比如告警或重试）。

方案2：用ExecuteProcess处理器（依赖系统命令） #

适合简单场景，需注意跨平台兼容性：

1. 添加ExecuteProcess处理器。
2. 根据操作系统配置命令：
   • Linux/macOS：命令设为mv，参数设为${in_progress_file_path} /path/to/completed/${filename}
   • Windows：命令设为move，参数设为"${in_progress_file_path}" "C:\path\to\completed\${filename}"
3. 配置关系：SUCCESS路由到正常流程，FAILURE路由到错误处理分支。

关键注意事项 #

• 确保NiFi运行用户对in_progress和completed目录有读写权限，否则会移动失败。
• 若业务逻辑处理器产生多分支FlowFile，需保证每个分支都携带in_progress_file_path属性，避免丢失路径无法执行移动。
• 可添加RouteOnAttribute处理器，在步骤7前检查in_progress_file_path属性是否存在，避免无效执行。

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