如何使用JoinEnrichment处理器实现两个JSON payload的流式合并？

前提准备 #

• 拥有两个JSON数据源：原始主数据流（需前置保留的内容）和Enrichment数据流（补充内容）
• 两个数据流需包含共同关联字段（如id），用于匹配待合并的对象

配置步骤 #

1. 预处理主数据流 #

在主数据流的上游处理器（如GetFile、ListenHTTP）中，通过EvaluateJsonPath提取关联字段并设置为FlowFile属性（例如record.id），或直接将JSON解析为NiFi的Record格式。

2. 配置JoinEnrichment核心参数 #

基础匹配规则

• Join Type：按需选择INNER_JOIN（仅保留两边匹配的记录）或LEFT_OUTER_JOIN（保留所有主数据流记录，无匹配时Enrichment字段留空）
• Join Strategy：选择MERGE_RECORDS，实现两个JSON对象的合并

关联字段映射

• Left Join Key(s)：填写主数据流的关联字段（如JSONPath路径/id，或对应FlowFile属性名）
• Right Join Key(s)：填写Enrichment数据流的对应关联字段

合并顺序与字段优先级

• Field Name Strategy：选择PREFER_LEFT_FIELD_NAMES，确保主数据流字段优先（重名字段保留主数据内容）
• Merge Direction：设置为LEFT_FIRST，保证合并后的JSON先包含主数据流所有字段，再追加Enrichment的非重复字段

3. 连接数据流 #

• 将主数据流接入JoinEnrichment的Left关系端口
• 将Enrichment数据流接入JoinEnrichment的Right关系端口
• 合并后的结果将从Success端口输出

示例验证 #

主数据流JSON：

{
  "id": "1001",
  "name": "Alice",
  "age": 30
}

Enrichment数据流JSON：

{
  "id": "1001",
  "email": "alice@example.com",
  "address": "123 Main St"
}

合并后输出JSON（原始+Enrichment顺序）：

{
  "id": "1001",
  "name": "Alice",
  "age": 30,
  "email": "alice@example.com",
  "address": "123 Main St"
}

注意事项 #

• 若存在重名字段，PREFER_LEFT_FIELD_NAMES会覆盖Enrichment侧字段；如需保留双方重名字段，可选择ADD_SUFFIX_TO_DUPLICATES，重名字段将自动添加后缀（如id_right）
• 确保两个数据流的Record Reader/Writer配置统一，均使用JSON格式处理器（如JsonTreeReader、JsonRecordSetWriter），避免合并失败

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