如何在NiFi中捕获Flow/流水线及对应处理器组的启动时间

NiFi 本身没有内置的处理器组启动时间戳属性，可通过以下两种方案实现需求： #

方案一：手动配置变量（无额外组件，适合低频次启动场景） #

该方案操作最简单，没有额外的处理器开销：

• 右键点击你的S3链路所属的处理器组，选择「变量」选项，新建名为pipeline_start_ts的变量，初始值可随意填写。
• 每次启动该处理器组前，先将pipeline_start_ts的值修改为当前时间戳（可直接填写毫秒级时间戳），保存变量配置后再启动处理器组即可。
• 组内所有ListS3、FetchS3、PutS3处理器均可直接通过表达式语言${pipeline_start_ts}调用该时间戳，比如你需要在PutS3的目标前缀中加入启动时间，直接配置为/your-target-prefix/${pipeline_start_ts}/即可生效。

方案二：自动生成时间戳（无需手动操作，适合高频启动场景） #

如果需要避免每次手动修改变量，可通过组内增加2个轻量处理器实现自动捕获：

• 第一步：在同处理器组内新增GenerateFlowFile处理器，配置如下：
  • 调度页签：运行策略选择「仅主节点运行」，调度间隔设置为9999 d（即9999天执行一次，保证仅处理器组启动时执行一次）
  • 属性页签：新增自定义属性pipeline_start_ts，值填写${now():toNumber()}，自动生成当前毫秒级时间戳
  • 自动终止关系：勾选success关系
• 第二步：新增UpdateAttribute处理器，配置「更新处理器组变量」规则，将流文件的pipeline_start_ts属性值同步到处理器组的同名变量中，自动终止所有关系。
• 第三步：将GenerateFlowFile与UpdateAttribute串联，再给ListS3、FetchS3、PutS3三个处理器设置3秒的初始启动延迟，保证变量更新完成后再运行S3链路。
• 后续每次启动处理器组，都会自动生成最新的启动时间戳同步到组变量，全组处理器均可正常调用。

补充说明 #

你可以根据业务需求用表达式语言调整时间戳格式，比如要转为年月日格式，直接调用${pipeline_start_ts:format("yyyyMMdd")}即可。 #

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