apache-nifi - 如何在 nifi 中为消息添加时间戳？

Question

免责声明：我对 nifi 一无所知。

我需要从ListenHTTP处理器接收消息，然后将每条消息转换为带时间戳的 json 消息。

所以，假设我在早上 5 点收到消息hello world。它应该将其转换为{"timestamp": "5 am", "message":"hello world"}.

我怎么做？

标签： apache-nifi

Answer 1

每个流文件都有属性，它们是存储在内存中键/值对中的元数据片段（可用于快速读/写）。当发生任何操作时，NiFi 框架都会将元数据片段写入与流文件相关的出处事件，有时还会写入流文件本身。例如，如果ListenHTTP是流中的第一个处理器，则任何进入流的流文件都将具有一个属性entryDate，该属性的值是它起源于 format 的时间Thu Jan 24 15:53:52 PST 2019。您可以使用各种处理器（即 、 等）读取和写入这些UpdateAttribute属性RouteOnAttribute。

对于您的用例，您可以在ReplaceText处理器之后立即ListenHTTP使用搜索值(?s)(^.*$)（整个流文件内容，或“您通过 HTTP 调用收到的内容”）和替换值{"timestamp_now":"${now():format('YYYY-MM-dd HH:mm:ss.SSS Z')}", "timestamp_ed": "${entryDate:format('YYYY-MM-dd HH:mm:ss.SSS Z')}", "message":"$1"}.

上面的示例提供了两个选项：

1. 这entryDate是流文件通过ListenHTTP处理器产生的时间
2. 该now()函数获取自纪元以来的当前时间戳（以毫秒为单位）

这两个值可能会因性能/排队/等而略有不同。在我的简单示例中，它们相隔 2 毫秒。您可以使用该format()方法和正常的Java 时间格式语法h a来格式化它们，因此您可以通过使用（完整示例：）来获得“5 am” now():format('h a'):toLower()。

例子

• ListenHTTP9999在带有路径的端口上运行contentListener
• ReplaceText如上
• LogAttribute带有日志有效负载true

卷曲命令：curl -d "helloworld" -X POST http://localhost:9999/contentListener

示例输出：

2019-01-24 16:04:44,529 INFO [Timer-Driven Process Thread-6] o.a.n.processors.standard.LogAttribute LogAttribute[id=8246b0a0-0168-1000-7254-2c2e43d136a7] logging for flow file StandardFlowFileRecord[uuid=5e1c6d12-298d-4d9c-9fcb-108c208580fa,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1548374015429-1, container=default, section=1], offset=3424, length=122],offset=0,name=5e1c6d12-298d-4d9c-9fcb-108c208580fa,size=122]
--------------------------------------------------
Standard FlowFile Attributes
Key: 'entryDate'
    Value: 'Thu Jan 24 16:04:44 PST 2019'
Key: 'lineageStartDate'
    Value: 'Thu Jan 24 16:04:44 PST 2019'
Key: 'fileSize'
    Value: '122'
FlowFile Attribute Map Content
Key: 'filename'
    Value: '5e1c6d12-298d-4d9c-9fcb-108c208580fa'
Key: 'path'
    Value: './'
Key: 'restlistener.remote.source.host'
    Value: '127.0.0.1'
Key: 'restlistener.remote.user.dn'
    Value: 'none'
Key: 'restlistener.request.uri'
    Value: '/contentListener'
Key: 'uuid'
    Value: '5e1c6d12-298d-4d9c-9fcb-108c208580fa'
--------------------------------------------------
{"timestamp_now":"2019-01-24 16:04:44.518 -0800", "timestamp_ed": "2019-01-24 16:04:44.516 -0800", "message":"helloworld"}