Kafka sink to InfluxDB

user_17329364 0
Viewed 228

I'm trying to get data from my kafka topic into InfluxDB using the Confluent/Kafka stack. At the moment, the messages in the topic have a form of {"tag1":"123","tag2":"456"} (I have relatively good control over the message format, I chose the JSON to be as above, could include a timestamp etc if necessary).

Ideally, I would like to add many tags without needing to specify a schema/column names in the future.

I followed https://docs.confluent.io/kafka-connect-influxdb/current/influx-db-sink-connector/index.html (the "Schemaless JSON tags example") as this matches my use case quite closely. The "key" of each message is currently just the MQTT topic name (the topic's source is an MQTT connector). So I set the "key.converter" to "stringconverter" (instead of JSONconverter as in the example).

Other examples I've seen online seem to suggest the need for a schema to be set, which I'd like to avoid. Using InfluxDB v1.8, everything on Docker/maintained on Portainer.

I cannot seem to start the connector and never get any data to move across.

Below is the config for my InfluxDBSink Connector:

{
  "name": "InfluxDBSinkKafka",
  "config": {
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",
    "name": "InfluxDBSinkKafka",
    "connector.class": "io.confluent.influxdb.InfluxDBSinkConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "topics": "KAFKATOPIC1",
    "influxdb.url": "http://URL:PORT",
    "influxdb.db": "tagdata",
    "measurement.name.format": "${topic}"
  }
}

The connector fails, and each time I click "start" (the play button) the following pops up in the connect container's logs:

[2022-03-22 15:46:52,562] INFO [Worker clientId=connect-1, groupId=compose-connect-group]
Connector InfluxDBSinkKafka target state change (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2022-03-22 15:46:52,562] INFO Setting connector InfluxDBSinkKafka state to STARTED (org.apache.kafka.connect.runtime.Worker)
[2022-03-22 15:46:52,562] INFO SinkConnectorConfig values: 
    config.action.reload = restart
    connector.class = io.confluent.influxdb.InfluxDBSinkConnector
    errors.deadletterqueue.context.headers.enable = false
    errors.deadletterqueue.topic.name = 
    errors.deadletterqueue.topic.replication.factor = 3
    errors.log.enable = false
    errors.log.include.messages = false
    errors.retry.delay.max.ms = 60000
    errors.retry.timeout = 0
    errors.tolerance = none
    header.converter = null
    key.converter = class org.apache.kafka.connect.storage.StringConverter
    name = InfluxDBSinkKafka
    predicates = []
    tasks.max = 1
    topics = [KAFKATOPIC1]
    topics.regex = 
    transforms = []
    value.converter = class org.apache.kafka.connect.json.JsonConverter
 (org.apache.kafka.connect.runtime.SinkConnectorConfig)
[2022-03-22 15:46:52,563] INFO EnrichedConnectorConfig values: 
    config.action.reload = restart
    connector.class = io.confluent.influxdb.InfluxDBSinkConnector
    errors.deadletterqueue.context.headers.enable = false
    errors.deadletterqueue.topic.name = 
    errors.deadletterqueue.topic.replication.factor = 3
    errors.log.enable = false
    errors.log.include.messages = false
    errors.retry.delay.max.ms = 60000
    errors.retry.timeout = 0
    errors.tolerance = none
    header.converter = null
    key.converter = class org.apache.kafka.connect.storage.StringConverter
    name = InfluxDBSinkKafka
    predicates = []
    tasks.max = 1
    topics = [KAFKATOPIC1]
    topics.regex = 
    transforms = []
    value.converter = class org.apache.kafka.connect.json.JsonConverter
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)

I am feeling a little out of my depth and would appreciate any and all help.

influxdb confluent-platform mqtt
1 Answers

The trick here is getting the data in the right format to Kafka in the first place. My MQTT source stream needed to have the value converter set to Bytearray with e schema url and schema = true. Then the Influx Sink started working when I used the jsonconverter, with schema=false. Then it started working. This is deceptive because the message queue looks the same with different valueconverters for the MQTT source connecter, so it took a while to figure out that was the problem.

After getting this working, and realising the confluent stack was perhaps a little overkill for this task, I went with the (much) easier route of pushing MQTT directly to Telegraf and having Telegraf push into InfluxDB. I would recommend this.

user_173293640
Related