如何在Groovy脚本中读取NiFi多FlowFile并区分、对比输出差异

核心思路 #

要处理两个FlowFile并区分它们，关键是先给两个文件打属性标记，再在Groovy脚本中批量获取并识别，最后逐行对比内容生成差异文件。

步骤1：给两个FlowFile添加区分属性 #

在ExecuteScript处理器之前，用UpdateAttribute分别给两个来源的FlowFile添加唯一属性：

• 第一个FlowFile添加属性：file.role=original
• 第二个FlowFile添加属性：file.role=comparison

如果两个FlowFile来自不同分支，可通过RouteOnAttribute分流后分别添加属性；如果是同一路径的成对文件，可结合MergeContent按组批量处理。

步骤2：Groovy脚本实现对比逻辑 #

在ExecuteScript处理器中使用以下代码，完成获取、识别、对比、输出的全流程：

// 1. 批量获取两个FlowFile
def flowFiles = []
def ff = session.get()
while (ff != null && flowFiles.size() < 2) {
    flowFiles.add(ff)
    ff = session.get()
}

// 校验是否拿到两个FlowFile，否则转移到失败关系
if (flowFiles.size() != 2) {
    flowFiles.each { session.transfer(it, REL_FAILURE) }
    return
}

// 2. 根据属性区分两个文件
def originalFile = flowFiles.find { it.getAttribute('file.role') == 'original' }
def comparisonFile = flowFiles.find { it.getAttribute('file.role') == 'comparison' }

// 校验属性是否存在，否则转移到失败关系
if (!originalFile || !comparisonFile) {
    flowFiles.each { session.transfer(it, REL_FAILURE) }
    return
}

// 3. 读取两个文件的内容行
def originalLines = []
session.read(originalFile) { inputStream ->
    originalLines = inputStream.readLines('UTF-8') // 按需修改编码
}

def comparisonLines = []
session.read(comparisonFile) { inputStream ->
    comparisonLines = inputStream.readLines('UTF-8')
}

// 4. 逐行对比生成差异内容
def differences = []
def maxLineCount = Math.max(originalLines.size(), comparisonLines.size())

for (int i = 0; i < maxLineCount; i++) {
    def origLine = i < originalLines.size() ? originalLines[i] : null
    def compLine = i < comparisonLines.size() ? comparisonLines[i] : null

    if (origLine != compLine) {
        if (origLine == null) {
            differences.add("+ 第${i+1}行: ${compLine}")
        } else if (compLine == null) {
            differences.add("- 第${i+1}行: ${origLine}")
        } else {
            differences.add("! 第${i+1}行内容不一致:")
            differences.add("  原文件: ${origLine}")
            differences.add("  对比文件: ${compLine}")
        }
    }
}

// 5. 生成差异输出文件
if (!differences.isEmpty()) {
    def diffContent = differences.join('\n')
    def outputFlowFile = session.create()
    outputFlowFile = session.write(outputFlowFile) { out ->
        out.write(diffContent.getBytes('UTF-8'))
    }
    // 给输出文件添加属性标记
    outputFlowFile = session.putAttribute(outputFlowFile, 'diff.status', 'has_differences')
    session.transfer(outputFlowFile, REL_SUCCESS)
}

// 处理原始文件，转移到成功关系（或按需转移到其他关系）
session.transfer(originalFile, REL_SUCCESS)
session.transfer(comparisonFile, REL_SUCCESS)

注意事项 #

• 编码设置：如果文件不是UTF-8编码，修改readLines和getBytes中的编码参数（如GBK）。
• 大文件处理：如果FlowFile内容极大，逐行读取会占用过多内存，可改用流式对比（逐行读取并即时对比，不存储所有行到内存）。
• 并发控制：将ExecuteScript的Concurrent Tasks设为1，确保每次只处理一对FlowFile，避免混乱。

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