我可以编写自定义 kafka 连接转换以将 JSON 转换为 AVRO 吗？

Question

我想使用 kafka-connect-hdfs 将无模式 json 记录从 kafka 写入 hdfs 文件。 如果我使用 JsonConvertor 作为键/值转换器，那么它不起作用。但是如果我使用的是 StringConvertor，那么它会将 json 写为转义字符串。

例如：

实际的 json -

{"name":"test"}

数据写入 hdfs 文件 -

"{\"name\":\"test\"}"

预期输出到 hdfs 文件 -

{"name":"test"}

有没有什么方法或替代方法可以实现这一点，或者我只能将它与模式一起使用？

以下是我尝试使用 JSONConvertor 时遇到的异常：

[2017-09-06 14:40:19,344] ERROR Task hdfs-sink-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:148)
org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.
    at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:308)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:406)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:180)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
    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:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

quickstart-hdfs.properties的配置：

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=test_hdfs_avro
hdfs.url=hdfs://localhost:9000
flush.size=1
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

connect-avro-standalone.properties的配置：

bootstrap.servers=localhost:9092
schemas.enable=false
key.converter.schemas.enable=false
value.converter.schemas.enable=false

Answer 1

当您在连接器的配置属性中指定转换器时，您需要包含与此转换器相关的所有属性，无论这些属性是否也包含在工作器的配置中。

在上面的示例中，您需要同时指定两者：

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

在快速启动-hdfs.properties中。

仅供参考，JSON 导出即将在 HDFS 连接器中推出。在此处跟踪相关的拉取请求：https://github.com/confluentinc/kafka-connect-hdfs/pull/196

更新：JsonFormat 已合并到 master 分支。