如何在Apache NiFi中基于CSV列值合并多个FlowFile？

1. 先把CSV转成结构化数据（两类FlowFile通用） #

用ConvertRecord处理器，把纯文本CSV转换成Avro/JSON格式的结构化记录，方便后续字段级操作：

• 配置Record Reader为CSVReader：设置Header Line Count为1，Quote Character为"，字段名自动读取表头即可。
• 配置Record Writer为AvroRecordWriter（性能更优）或JSONRecordWriter，输出结构化FlowFile。

2. 缓存第二类FlowFile的关联数据（支持跨多个文件匹配） #

要实现第一类FlowFile匹配所有同结构第二类FlowFile的d列，得先把第二类数据统一存入缓存：

• 用DistributedMapCachePut处理器：
  • 设置Cache Key为${record:d}（通过record路径提取d列值作为缓存键）。
  • 设置Cache Value为JSON格式的e/f/g字段值，比如{"e":"${record:e}","f":"${record:f}","g":"${record:g}"}，方便后续读取解析。
  • 多个第二类FlowFile会自动更新缓存，相同d值会覆盖旧数据（按需调整缓存策略）。

3. 第一类FlowFile关联缓存数据 #

用DistributedMapCacheGet处理器，基于d列值从缓存中拉取对应的e/f/g数据：

• 设置Cache Key为${record:d}，和缓存写入的键保持一致。
• 匹配成功的FlowFile会带上map.cache.value属性，未匹配的（比如d=456）该属性为空。

4. 拼接字段生成目标CSV #

用UpdateRecord处理器，把第一类字段和缓存获取的字段拼接：

• 配置Record Reader为之前的Avro/JSON Reader，Record Writer为CSVRecordWriter。
• 输出字段映射规则：
  • a: ${record:a}
  • b: ${record:b}
  • c: ${record:c}
  • d: ${record:d}
  • e: ${map.cache.value:e ?: ""}（缓存无匹配时用空字符串）
  • f: ${map.cache.value:f ?: ""}
  • g: ${map.cache.value:g ?: ""}
• CSVWriter设置Header Line Count为1，自动生成目标表头，Quote Character设为"。

误区说明 #

• MergeContent只适合同结构文件的简单合并，无法做字段级关联匹配，所以用它达不到需求。
• Fork/Join Enrichment默认是一对一的FlowFile匹配，不支持跨多个FlowFile的全局d列匹配，这也是你操作后出错的核心原因，改用分布式缓存才能实现全局匹配逻辑。

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