NiFi，与KafkaConsumer一起流写为json答案

【问题标题】：NiFi, flow with KafkaConsumer to write as jsonNiFi，与KafkaConsumer一起流写为json
【发布时间】：2018-01-24 17:34:44
【问题描述】：

目前我遇到以下问题：我正在使用 KafkaConsumer 阅读来自 Kafka 主题的消息。消息是字符串，格式如下： { "a" : "b", "a1" : "b1", "c2" : "c3" } 它们保存在 FlowFile 的有效负载中。

我想将该字符串转换为 json 或理想情况下转换为 csv，但不知道该怎么做。

我是 NiFi 的新手，并且尽可能地进行了研究，但我发现的答案是关于从 json 到 avro 或类似的转换，但从不将字符串转换为 json 或 avro。我还发现 Kafka 消息在 FlowFile 的有效负载中，而不是在属性中，所以我不知道如何获得它，因为示例总是涉及属性。

简而言之：我可以使用一些内置处理器将 FlowFile 的有效负载（一个字符串）转换为 json/cvs。

【问题讨论】：

【解决方案1】：

如果您的消息在 FlowFile 中，以下顺序可能会有所帮助：

1) 使用 AttributesToJson 将有效负载消息转换为 Json。 2) 使用 EvaluateJsonPath 提取有效负载消息。在你的情况下，卡夫卡消息。然后您可以将提取的消息传递给 csv 生成。

这篇文章可以帮助将 Json 转换为 CSV：Convert Json To CSV

【讨论】：

【解决方案2】：

我最终这样做了：

{ "a" : "b", "a1" : "b1" }

a -> $.a //results in attribute named a with value b

a1 -> $.a1 //results in attribute named a1 with value b1

Replacement value -> ${'a'},${'a1'}

结果为单行，但没有新行：

b,b1

添加新行追加\n, '\n', "\n" 没有 > 工作。 有效的方法是在替换值字段中键入时按Shift+Enter，这会导致创建一个空的新行。

【讨论】：