NiFi中MergeContent合并不同类型JSON流文件的配置问题

问题分析 #

当前MergeContent处理器只会合并同类型FlowFile，核心原因是分组属性配置错误，导致同类型文件被归为一组。需要调整配置逻辑，确保一个Create类型和一个Update类型的FlowFile被分到同一组，最终合并为指定的Transaction结构。

具体步骤 #

1. 给FlowFile添加分组与类型标识 #

使用UpdateAttribute处理器，为两个FlowFile添加属性：

• 给需要合并到同一事务的两个FlowFile设置完全相同的transaction.id属性（比如transaction.id=txn-20240520，可通过GenerateFlowFile生成唯一ID，保证同一组的两个文件属性值一致）
• 给Create类型的FlowFile添加content.type=Create，Update类型的添加content.type=Update（用于后续校验，可选）

2. 重新配置MergeContent处理器 #

修改以下关键属性：

• Correlation Attribute Name：设置为transaction.id，确保只有拥有相同该属性值的FlowFile会被纳入同一合并组
• Minimum Number of Entries：设为2，必须凑够2个FlowFile才触发合并
• Maximum Number of Entries：设为2，避免合并多余的FlowFile
• Merge Strategy：选择Concatenate
• Header：填入以下JSON片段作为输出开头：

  {"Transaction": {"Action": "Transaction", "Type": {
  

• Footer：填入以下JSON片段作为输出结尾：

  }}}
  

• Delimiter：设为逗号（,），用于分隔两个FlowFile的内容

3. 验证合并结果 #

完成配置后，两个FlowFile会被合并为你期望的结构：

{
  "Transaction": {
    "Action": "Transaction",
    "Type": {
      "Create": [
        {
          "Action": "Create",
          "Book": {
              "Id": "1234"
           }
        },
        {
          "Action": "Create",
          "Video": {
              "Id": 3245
          }
        }
      ],
      "Update": [
        {
          "Action": "Update",
          "Book": {
              "Id": "5376"
           }
        },
        {
          "Action": "Update",
          "Video": {
              "Id": "8267"
          }
        }
      ]      
    }
  }
}

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