如何使用Nifi将数据从Kafka传输到Cassandra？

Question

我想在Kafka中使用Nifi从Cassandra收集数据。我为this.创建了这样的流程

我的数据库连接配置如下：

这是我对ConvertJsonToSQL处理器的配置：

我在ConvertJsonToSQL处理器上遇到以下错误。

ConvertJSONToSQL [id = d25a7e27-0167-1000-2d9a-2c969b33482a]由于null，ConvertJSONToSQL [id = d25a7e27-0167-1000-2d9a-2c969b33482a]无法处理会话;处理器在行政上屈服1秒：java.lang.NullPointerException

注意：我将dbschema驱动程序jar添加到Nifi库中。

您认为我应该怎么做才能解决这个问题？

Answer 1

根据可用信息很难对错误进行故障排除，ConvertJSONToSQL失败的最可能原因是无效的JSON。距离documentation只有一点：

传入的FlowFile应该是“平坦的”JSON消息，这意味着它由单个JSON元素组成，每个字段映射到一个简单类型。

我无法看到你在AttributesToJSON处理器中做了什么，但我相信twitter通常会返回一个嵌套的JSON，并且你可能没有把它弄平。

---

解决此问题的一种简单通用方法是从顶部启动处理器，并在每个处理器之前/之后检查队列，直到看到您不期望的内容。

有了这个，您应该能够准确地找出问题，如果需要，您可以使用以这种方式发现的信息来创建可重现的示例并提出更详细的问题。