Nifi中FetchS3Object事件驱动获取S3文件的方案咨询

针对S3文件落地晚于调度时间导致拉取失败的问题，可选择以下两种方案实现适配：

方案1：改造为事件驱动模式（生产环境优先推荐） #

该方案完全匹配文件实际落地时间，无无效资源消耗：

• 先在目标S3 Bucket配置事件通知规则，触发条件选择 s3:ObjectCreated:*系列事件，事件目标绑定到AWS SQS队列
• NiFi侧将原有时间驱动的触发逻辑替换为ListSQS Processor作为流程起点，配置秒级间隔拉取SQS中的对象创建事件通知
• 从SQS消息体中解析出目标S3桶名、对象键、版本号等参数，作为属性传递给下游的FetchS3Object Processor执行文件拉取
• 可额外配置SQS消息去重、NiFi侧DistributedMapCache记录已拉取对象键，避免重复拉取同一份文件

方案2：配置循环重试直到拉取成功 #

如果暂不希望调整整体触发逻辑，可通过重试组件实现按需循环拉取：

• 调整原有时间调度组件仅单次生成携带目标S3桶名、对象键属性的FlowFile，将FetchS3Object的调度策略改为事件驱动（仅当有输入FlowFile时才执行）
• 在FetchS3Object的failure关系后接入RetryFlowFile Processor，按业务需求配置重试规则：
  • Maximum Retries：最大重试次数，可按业务允许的最大等待时长设置，例如等待24小时、每小时重试1次可设置为24
  • Penalization Period：两次重试的间隔时间，按实际业务容忍的延迟设置即可
• 将RetryFlowFile的Retry关系路由回FetchS3Object的输入队列，Retries Exceeded关系可路由到告警组件或终止队列，避免无限重试
• 可选优化：在重试链路中加入RouteOnAttribute Processor，判断当前时间是否超出业务预设的文件最晚落地阈值，超出后直接终止流程

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