Kafka Connect Sink-来自：Avro Topic，到：Json->Redis

Question

我有一个环境，我使用一个Kafka Connect Worker，它使用Oracle数据库中的一些数据，然后将其推送到Avro格式的Kafka主题中。

现在，我需要创建一个Kafka连接接收器来使用这个AVRO消息，将其转换为Json，然后将其写入Redis数据库。

到目前为止，我只能在Redis上写我在topic中使用的同样的AVRO消息。我曾尝试使用转换器，但可能误解了其用法。

吼我的工人和水槽的配置。

{
    "name": "SOURCE",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "transforms": "createKey, extractStr",
        "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
        "transforms.createKey.fields": "ID",
        "transforms.extractStr.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.extractStr.field": "ID",
        "connection.url": "<>",
        "connection.user": "<>",
        "connection.password": "<>",
        "table.whitelist": "V_TEST_C",
        "schema.pattern": "<>",
        "numeric.mapping": "best_fit",
        "mode": "timestamp+incrementing",
        "incrementing.column.name": "CID",
        "timestamp.column.name": "TS_ULT_ALT",
        "validate.non.null": "false",
        "table.types": "VIEW",
        "retention.ms":12000,
        "poll.interval.ms": "30000",
        "topic.prefix": "TEST.",
        "value.converter.schema.registry.url": "<>"
    }
}

水槽

{
  "name": "SINK",
  "config": {
    "connector.class": "com.github.jcustenborder.kafka.connect.redis.RedisSinkConnector",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "tasks.max": "1",
    "topics": "V_TEST_C",
    "redis.hosts": "redis:6379",
    "schema.registry.url": "<>",
    "value.converter.schema.registry.url": "<>",
    "value.converter.schemas.enable":"false",
    "key.converter.schemas.enable":"false",
    "insert.mode": "UPSERT",
    "delete.enabled": "false",
    "quote.sql.identifier": "never"
  }
}

Answer 1

在Avro格式的Kafka主题中推出

那不是你贴的

"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",

在任何情况下，都不能使用Redis连接器存储除字符串或字节之外的任何内容

来自文档（强调添加）

此连接器期望来自Kafka的记录具有存储为字节或字符串的键和值。如果您的数据已经以您希望在Redis中使用的格式在Kafka中，请考虑为此连接器使用ByteArrayConverter或StringConverter

因此，使用AvroConverter是不可能的，因为它生成的是结构化对象，而不是字符串或字节的模式。

注意：禁用模式的JSONConverter实际上与处理JSON数据的字符串转换器的行为相同。然而，如果将StringConverter与Avro数据一起使用，最终可能会得到类似于Struct{foo=bar}的数据，但您仍然需要在某种程度上运行Avro反序列化器，因此我不确定您是否真的可以将Json或StringConverter与Avro主题数据一起使用。