如何配置Apache NiFi的PutHiveStreaming实现全量提交无重复写入？

1. 基于Records per Transaction的事务控制 #

你的思路是可行的，通过配置Records per Transaction为FlowFile的总记录数，就能实现单个FlowFile内数据的“全有或全无”写入。具体操作如下：

• 先用ExtractAvroMetadata或CountRecord处理器统计当前FlowFile的Avro记录总数，把结果存入自定义属性（比如total.record.count）。
• 在PutHiveStreaming处理器中，将Records per Transaction的值设为${total.record.count}（通过表达式语言读取属性值）。

这样配置后，NiFi会把整个FlowFile的所有记录打包成一个Hive Streaming事务：只有所有记录都成功写入并提交，事务才完成，FlowFile标记为成功；如果中途出现节点宕机、任务中断等异常，整个事务会被终止，Hive不会接收任何部分数据。同时因为你关闭了Rollback on Failure，失败的FlowFile会直接路由到失败关系，不会阻塞后续数据流入。

2. 解决重试导致的重复数据问题 #

NiFi的重试机制会重新提交未成功处理的FlowFile，这确实可能造成Hive表出现重复数据，可通过以下两种方式解决：

方式一：Hive端实现幂等写入 #

• 如果你的Hive表支持ACID（需开启Hive ACID配置，使用ORC格式），可以将PutHiveStreaming的写入模式改为Upsert，结合表中的唯一业务键（比如用户ID+数据生成时间戳），重复数据写入时会自动覆盖或忽略。
• 对于非ACID表，可在写入前用INSERT ... WHERE NOT EXISTS语句过滤已存在的记录，但这种方式会增加额外查询操作，带来一定延迟。

方式二：NiFi端提前去重 #

• 使用DistinctRecord处理器，基于记录的唯一标识字段（比如业务主键）对FlowFile内的记录去重。
• 借助NiFi自带的DistributedMapCache：写入前先将每条记录的唯一标识存入缓存；写入时查询缓存，仅处理未存在的记录；事务提交成功后，将标识永久存入缓存；若事务失败，则删除缓存中的临时标识。

3. 额外配置注意事项 #

• 确保Batch Size参数与Records per Transaction保持一致，避免NiFi自动拆分记录批次。
• 给失败的FlowFile设置合理的Max Retries次数，避免无限重试占用资源。
• 定期清理失败队列中的无效FlowFile，防止队列积压。

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