我正在尝试使用Confluent InfluxDB Sink Connector将kafka主题中的数据导入到我的InfluxDB中。
首先,我使用nifi从日志文件向kafka主题传输数据,效果很好。 kafka主题获取数据,如下所示:
{
"topic": "testDB5",
"key": null,
"value": {
"timestamp": "2019-03-20 01:24:29,461",
"measurement": "INFO",
"thread": "NiFi Web Server-795",
"class": "org.apache.nifi.web.filter.RequestLogger",
"message": "Attempting request for (anonymous)
},
"partition": 0,
"offset": 0
}
然后,我通过Kafka Connect UI创建InfluxDB接收器连接器,我得到以下异常:
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:323)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.NullPointerException
at io.confluent.influxdb.InfluxDBSinkTask.put(InfluxDBSinkTask.java:140)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:565)
... 10 more
但是,如果我通过使用手动输入数据到另一个主题testDB1
./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic testDB1 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"measurement","type":"string"},{"name":"timestamp","type":"string"}]}'
它工作,我的流量数据库可以获取数据。
这是连接配置:
connector.class=io.confluent.influxdb.InfluxDBSinkConnector
influxdb.url=http://myurl
tasks.max=1
topics=testDB5
连接主题testDB1的配置与主题名称相同。
在nifi有什么问题吗?但它可以很好地将数据传输到主题。
当您将Avro与Kafka Connect一起使用时,Avro deserialiser希望使用Avro serialiser对数据进行序列化。这就是kafak-avro-console-producer
使用的,这就是你使用它时管道工作的原因。
This article为Avro和Schema Registry提供了良好的背景。另见Kafka Connect Deep Dive – Converters and Serialization Explained。
我不熟悉Nifi,但看看文档,AvroRecordSetWriter似乎可以选择使用Confluent Schema Registry。猜测你也想把Schema Write Strategy
设置为Confluent Schema Registry Reference
。
一旦您可以使用kafka-avro-console-consumer
从主题中使用数据,那么您就知道它已正确序列化并且可以与Kafka Connect接收器一起使用。
我找到了原因。这是因为在Nifi中,我使用PublishKafka_0_10
将数据发布到Kafka主题,但它的版本很低!
当我在ksql中进行查询时,它会说
Input record ConsumerRecord(..data..) has invalid (negative) timestamp.
Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp,
or because the input topic was created before upgrading the Kafka cluster to 0.10+. Use a different TimestampExtractor to process this data.
所以,我把它改成PublishKafka_1_0
,然后重新开始,它有效!我的流入数据库可以获取数据。我无语了。
感谢Robin Moffatt的回复,这对我很有帮助。