记录富集用NiFi流

Question

我正在使用NIFI1.11.4构建一个数据管道，其中IoT设备以JSON格式发送数据。每次从IoT设备接收数据时，都会收到两个JSONS；

JSON_Initial

{
   devId: "abc",
   devValue: "TWOINITIALCHARS23",
}

和JSON_FINAL

{
   devId: "abc",
   devValue: "TWOINITIALCHARS45",
}

{
   devId: "abc",
   devValue: "2345",
}

Answer 1

假设devid对于设备是静态的，并且不用于相关（即对于来自该设备的所有消息，abc，对于前两个消息不是abc，然后对于后两个消息则是def，等等），您有以下几个选项：

• 使用mergeContent将flowfile内容（两个JSON块）串联起来，使用replaceText修改组合的内容以匹配所需的输出。这将需要调整MC binning属性，将合并窗口限制在1-2秒内（例如，如果每秒接收多条消息，这是困难的/不够的），并使用正则表达式删除重复内容。
• 使用自定义脚本与设备JSON输出交互（例如，Groovy将使JSON交互非常简单）
  • 如果您在NiFi上下文中（通过executescript或invokescriptedprocessor)执行此操作，您将可以访问NiFi框架，因此您可以计算流文件属性和内容，从而使此操作更加容易（将有初始时间戳等属性）。
  • 如果您在NiFi上下文之外（通过executeProcess或executeStreamCommand)执行此操作，您将无法访问NiFi框架（属性等），但您可以直接与设备进行更好的交互。