NiFi - ExecuteSQLRecord 处理器查询中使用的持久时间戳值答案

【问题标题】：NiFi - Persist timestamp value used in ExecuteSQLRecord processor queryNiFi - ExecuteSQLRecord 处理器查询中使用的持久时间戳值
【发布时间】：2020-10-28 13:44:20
【问题描述】：

我的用例很简单，但到目前为止我还没有找到合适的解决方案。

我编写查询，在 ExecuteSQLRecord 处理器命中时用列中的当前时间戳标记数据，并从数据库中获取数据现在我想要的是创建的流文件也必须在他的名字中具有相同的时间戳但我不知道如何捕获 ${now():format("yyyyMMddHHmmss")} 的属性，所以我可以使用 alter 重命名流文件

基本上，我想“在我访问数据库时”存储时间戳，我不能在 executeSQL 处理器之前使用更新处理器来获取所需的时间戳（为什么 => 因为如果之前的执行仍在进行中使用 executeSQL 并且所有流文件将通过带有时间戳值的 updateattribute 处理器，并将位于队列中，直到 executeSQL 处理器处理当前线程。

注意 - 我在独立模式下运行 NiFi，所以我不能在多个线程中运行 executeSQL。

非常感谢任何帮助。提前致谢

【问题讨论】：

您想将属性值放入流文件中吗？
是的，使用我在查询中完成的时间戳更新流文件，但是当我想使用相同的时间戳值来更新相同的流文件名称时，第二部分很棘手
如果您只想将属性放在内容中，请尝试使用处理器 ReplaceText 并将您的属性放在替换字段中。如果您只想在内容末尾收集此信息，则可以选择附加模式。
必须将时间戳添加到我的流文件内容的每一行中，我在流文件中有 csv 格式，并且该时间戳列显示了处理时间，我想稍后在文件名中使用它流文件
在这种情况下，您可以使用带有正则表达式替换选项的 ReplaceText。

标签： apache-nifi

【解决方案1】：

ExecuteSQLRecord 写入一个名为 executesql.query.duration 的属性，其中包含查询的持续时间 + 获取时间（以毫秒为单位）。

因此，我们可以在使用${now():toNumber():minus(${executesql.query.duration})} 的 ExecuteSQLRecord 之后放置一个 UpdateAttribute 处理器，以获取当前时间为 Epoch Millis，然后减去总查询持续时间，以获取查询开始的时间。

然后您可以使用:format('yyyyMMddHHmmss') 将其恢复为您想要的时间戳格式。

可能与准确时间相差几毫秒（到达 UpdateAttribute 处理器所用的时间）。

See docs for ExecuteSQLRecord

【讨论】：