我对 NiFi 很陌生。我们已经完成了可以使用 Kafka 消息的设置。 在 NiFi UI 中，我使用 ConsumeKafka_0_10 创建了处理器。当消息发布（不同的进程）时，我的处理器能够正确地获取所需的数据/消息。 我转到“数据来源”，可以看到收到了正确的数据。
但是，我希望将下一个流程作为验证者。这将从 consumekafka 读取流文件并进行基本验证（用户提供的脚本应该很好）
我们如何使用那个或哪个处理器在这里工作？
还有什么方法可以将流文件输入格式转换为 csv 或 json 格式？
【问题讨论】：
您有几个选择。根据流文件内容格式，您可以将ValidateRecord 与配置为验证它的*Reader 记录读取器控制器服务一起使用。如果您已经在 Groovy/Javascript/Ruby/Python 中有一个脚本来执行此操作，ExecuteScript 也是一个解决方案。
同样，要将流文件内容转换为 CSV 或 JSON，请使用 ConvertRecord 处理器、ScriptedReader 和 CSVRecordSetWriter 或 JsonRecordSetWriter 以输出为正确的格式。这些进程在内部使用 Apache NiFi 记录结构从任意输入/输出格式转换为高性能。更多阅读请访问blogs.apache.org/nifi 和bryanbende.com。
【讨论】：