NiFi中PutFile与ExecuteScript间流文件滞留重复处理问题求助

核心问题定位 #

你的ExecuteScript脚本逻辑存在错误：每次调度都主动创建新流文件，完全忽略了上游PutFile传递过来的流文件。这会引发两个关键问题：

• 上游PutFile输出的流文件因未被下游处理器接收处理，持续滞留在连接中，触发重复调度或重试
• ExecuteScript按调度间隔（4秒）持续生成新流文件，即便设置了流文件过期时间，也会因每次调度都生成新实例，导致重复发送至API

修复步骤 #

1. 修改ExecuteScript脚本，正确处理上游流文件 #

删除session.create()，改为获取上游传递的流文件，脚本调整为：

import org.apache.commons.io.IOUtils
import org.apache.nifi.processor.io.OutputStreamCallback

// 获取上游传递的流文件，而非创建新文件
def flowFile = session.get()
if (!flowFile) return

def filePath = 'E:/tmp/tmp_file.txt'
def file = new File(filePath)

if (file.exists()) {
    def fileContent = IOUtils.toByteArray(new FileInputStream(file))
    // 将内容写入上游流文件
    flowFile = session.write(flowFile, { outputStream ->
        outputStream.write(fileContent)
    } as OutputStreamCallback)
    session.transfer(flowFile, REL_SUCCESS)
} else {
    // 文件不存在时，移除流文件并传递至失败关系
    session.remove(flowFile)
    session.transfer(REL_FAILURE)
}

2. 调整处理器调度与流文件配置 #

• ExecuteScript调度策略：改为Event-Driven（事件驱动），仅在有上游流文件到达时执行，避免空调度生成无效流文件
• 流文件过期时间：保持1秒或按需调整，同时确保PutFile的Penalty Duration（惩罚时长）不与过期时间冲突
• PutFile配置：检查Conflict Resolution Strategy（冲突解决策略），避免因文件写入冲突导致流文件重试

3. 排查连接队列滞留原因 #

• 查看连接的Back Pressure（背压）配置，确认队列未达到背压阈值导致流文件堆积
• 为ExecuteScript的REL_FAILURE关系配置后续处理，避免失败流文件长期滞留队列

额外优化建议 #

• 若业务仅需读取本地文件内容到流文件，建议用FetchFile处理器替代ExecuteScript，无需自定义脚本，稳定性更高
• 启用NiFi的FlowFile Provenance（溯源）功能，查看流文件生命周期，精准定位重复处理的节点

内容的提问来源于stack exchange，提问作者Nofal Al-Mukaddam