CRON驱动下NiFi的FlowFile全量摄入及属性去重方案咨询

嘿，我来帮你搞定这两个NiFi的需求，分两部分详细说：

这可以通过「排序+去重」的组合处理器实现，步骤很清晰：

• 第一步：给FlowFile按日期排序
  先用SortRecord处理器，配置排序规则：

  • 主排序字段选product_code，类型设为String，排序方向Ascending（升序），这样相同product_code的FlowFile会被归为一组；
  • 次排序字段选publication_date，类型设为Date，格式匹配你的yyyy-MM-dd，排序方向Descending（降序），确保每个product_code下最新的日期排在最前面。

• 第二步：去重保留最新项
  接着用DetectDuplicate处理器，做如下关键配置：

  • Duplicate Detection Strategy选择DETECT_DUPLICATES_BASED_ON_ATTRIBUTES；
  • Attributes to Compare填写product_code，只基于这个属性判断重复；
  • 勾选Keep Only Unique FlowFiles并设为true；
  • 勾选First Unique Wins并设为true。
    因为已经按日期降序排过，每个product_code的第一个FlowFile就是最新的，后面的都会被标记为重复并过滤掉，刚好满足你的需求。

核心是确保同一CRON批次的FlowFile全部到达后再处理，同时每次批处理的去重状态独立：

• 方法一：批次标记+等待全量FlowFile

  1. 在CRON触发的起始处理器（比如生成FlowFile的节点）后，添加UpdateAttribute，给每个FlowFile新增batch_id属性，值用${now():format("yyyyMMdd")}（和每日CRON周期匹配），让当天所有FlowFile共享同一个批次ID。
  2. 接入Wait处理器，配置：
     • Wait Strategy选Wait for Signal；
     • Signal Identifier和Release Signal Identifier都填${batch_id}。
  3. 同时用ExecuteScript或RouteOnAttribute配合DistributedMapCachePut统计同一batch_id的FlowFile数量，当数量达到预期批次总量时，触发Wait的信号释放，让所有FlowFile同时进入DetectDuplicate。

• 方法二：重置DetectDuplicate状态（更简洁的批处理方案）
  因为你是每日独立批处理，不需要和历史批次关联：

  1. 在CRON流程的最开头，添加ResetProcessorState处理器，指定目标为你的DetectDuplicate节点，每次批处理开始前自动清空它的去重状态。
  2. 之后让所有FlowFile自然流入DetectDuplicate即可——状态清空后，它会把同一product_code的FlowFile中第一个（也就是排序后的最新项）保留，其余标记为重复。

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