如何在Apache NiFi中将SplitJson拆分的JSON记录存入变量

你已经通过SplitJson完成数组拆分的前提下，按以下步骤配置即可，根据后续使用场景分两种情况处理： #

前置通用配置（所有场景都要做） #

SplitJson处理器拆分后会自动给每个输出FlowFile打3个分片内置属性，不需要额外生成序号：

• fragment.identifier：同一次拆分生成的所有FlowFile共享同一个ID，用来区分不同批次的拆分结果
• fragment.index：当前分片的序号，从0开始计数，对应第一个FlowFile值为0、第二个为1、第三个为2
• fragment.count：当前拆分批次的总分片数，这里固定为3

1. 把SplitJson的成功关系连接到UpdateAttribute处理器，新增1个动态属性：
   • 属性名：variableName
   • 属性值：variable${fragment.index + 1}
     配置后三个FlowFile会分别得到variableName值为variable1、variable2、variable3
2. 把UpdateAttribute的成功关系连接到ExtractText处理器，新增1个捕获规则：
   • 属性名：fileContent
   • 正则规则：(?s)(^.*$)
     开启正则的DOTALL单行匹配模式，就能把当前FlowFile的完整JSON内容捕获到fileContent属性中，不会丢失转义字符或换行。

场景1：每个FlowFile单独使用对应编号变量 #

如果下游节点只需要处理单个FlowFile，只要拿到当前FlowFile对应的编号变量即可，不需要三个变量同时存在：

• 在ExtractText后再接一个UpdateAttribute处理器，新增动态属性时，点击属性名输入框右侧的表达式语言开关，开启属性名的表达式解析能力，属性名填${variableName}，属性值填${fileContent}
• 配置完成后，三个FlowFile会分别自带对应属性：
  • FlowFile1带variable1属性，值为自身完整JSON内容
  • FlowFile2带variable2属性，值为自身完整JSON内容
  • FlowFile3带variable3属性，值为自身完整JSON内容
• 直接把FlowFile送到下游处理器即可，下游可以直接通过${variable1}/${variable2}/${variable3}表达式调用对应值（每个FlowFile只会带自己对应的那个变量属性）。

场景2：三个变量同时存在于同一个FlowFile供统一调用 #

如果后续节点需要同时拿到三个JSON内容做统一处理，需要把三个分片的内容聚合到同一个FlowFile上，用分布式缓存实现最稳定：

1. 先在NiFi全局配置里添加1个DistributedMapCacheServer控制器服务，端口用默认的4557即可，启动服务；再添加1个DistributedMapCacheClientService，关联刚才的缓存服务，启动。
2. 把前置配置中ExtractText的成功关系连接到PutDistributedMapCache处理器：
   • 关联刚才配置好的缓存客户端
   • 缓存Entry Key设置为${fragment.identifier}_${variableName}
   • 缓存Entry Value设置为${fileContent}
     这一步会把三个FlowFile的内容按批次ID_变量名的规则存到缓存里。
3. 把PutDistributedMapCache的成功关系连接到RouteOnAttribute处理器，新增路由规则：
   • 规则名：lastFragment
   • 匹配条件：${fragment.index == fragment.count - 1}
     这一步会把每个拆分批次的最后一个FlowFile（也就是第三个FlowFile）单独路由出来，作为聚合后的载体FlowFile。
4. 把lastFragment路由关系连接到FetchDistributedMapCache处理器，关联同一个缓存客户端，新增三个抓取规则：
   • 缓存Key：${fragment.identifier}_variable1，存入属性名：variable1
   • 缓存Key：${fragment.identifier}_variable2，存入属性名：variable2
   • 缓存Key：${fragment.identifier}_variable3，存入属性名：variable3

配置完成后，输出的FlowFile会同时携带variable1/variable2/variable3三个属性，值分别对应三个拆分后FlowFile的完整JSON内容，可以直接供后续节点统一调用。

注意：不建议把动态生成的流内容存为进程组级别的全局变量，全局变量是静态配置，适合存固定的连接参数、阈值这类值，流内动态传递内容用FlowFile属性是官方推荐的最优方案。

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