NiFi 如何在 UpdateAttribute 处理器中获取更新的 JSON 值？

Question

我有以下工作流程：

ConsumeKafkaData->EvaluateJsonFlow->UpdateRecord->Putfile.

我收到以下简单的 JSON 消息： {"name":"Yeshwant","active":false}

在 EvaluateJsonPath 中，我在配置中创建了一个属性“name”，该属性链接到上述 json 数据的“name”属性。

在 UpdateAttribute 处理器中，我创建了一条规则，当“name”的值等于“dummy”时，将更改为“funny”。我能够将属性“name”的值更改为“funny”。但是输入输出的json数据没有变化。为什么？ UpdateAttribute 如何直接更改 JSON 中的数据？有没有我遗漏的配置？

即使在 PutFile 处理器中，我也没有看到 json 数据发生变化。

谢谢 好想

Answer 1

在现有流程中，您更改的是 attribute associated to the flowfile 的值，而不是流程文件的实际内容。

在 UpdateAttribute 处理器使用 AttributesToJson 处理器并重新创建输出 JSON 内容之后。

流程：

1. ConsumeKafkaData
2. EvaluateJsonFlow //Extract all json data attributes as flowfile attributes
3. UpdateAttribute 
4. AttributeToJSON //AttributesList -> name,active; Destination -> flowfile Content
5. Putfile

有关 AttributesToJSON 处理器的更多详细信息/用法，请参阅this 链接。

此外，您可以在这种情况下使用面向记录的处理器，即 UpdateRecord/QueryRecord 处理器，然后使用 LiteralValue策略并在name 属性的值上使用Expression language if-else。

如果你使用QueryRecord processor，那么在你的sql查询中写一个case语句。

有关更多详细信息/使用 updateRecord 处理器，请参阅 this 链接。

有关 QueryRecord 处理器的更多详细信息/用法，请参阅 this 链接。