NiFi两组数据过滤合并及等待机制相关技术咨询

1. NiFi是否支持此类操作？ #

完全支持。NiFi的核心设计就是为了灵活的数据流编排，具备路由、属性提取、状态管理、批量数据处理等能力，完全能实现分组、ID交叉匹配过滤、数据对齐这类需求。

2. 可使用哪些处理器实现该对比过滤？ #

根据不同的场景（实时/批量），可以选择以下处理器组合：

• ID提取阶段：
  • ExtractText：从文本格式数据中提取ID，存入FlowFile属性。
  • JsonPathReader/XmlXPathReader：针对JSON/XML格式数据，用路径表达式提取ID到属性。
• 对比过滤阶段：
  • DistinctHashLookup：维护一个内存哈希表，先让一组数据（比如A组）写入哈希表存储ID，另一组（B组）数据过来时查询ID是否存在，实现双向过滤。支持设置ID过期时间，避免内存占用过高。
  • MergeContent + RouteOnAttribute：先将A组所有FlowFile的ID合并为一个集合（如JSON数组），存入属性；然后在RouteOnAttribute中用EL表达式${id:notIn(${a_group_ids})}判断B组ID是否不在A组集合中，反之同理，实现过滤。
  • DistributedMapCacheClientService + RouteOnAttribute：借助分布式缓存存储A/B组的ID集合，后续FlowFile通过缓存查询ID是否存在，适合大规模数据场景。

3. 如何保证对比时A、B两组数据均已完成等待？ #

可以通过**Wait+Notify的多信号机制**实现，步骤如下：

• 给A、B组分别设置独立的完成信号，比如group_a_complete和group_b_complete。
• 对A组：用Count处理器统计该组FlowFile数量，当达到预期总数时，触发Notify处理器发送group_a_complete信号；同理处理B组，发送group_b_complete信号。
• 在后续的Wait处理器中，设置等待多个信号，将两个信号名称都填入等待列表，只有当两个信号都被接收后，才释放FlowFile进入对比环节。
• 额外方案：用DistributedMapCacheClientService存储两组的完成状态，A组完成后写入缓存group_a:done，B组写入group_b:done；然后用RouteOnAttribute循环检查缓存中两个状态是否都为done，直到满足条件再进入下一步。

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