Kafka Connect HDFS Sink 使用 JsonConverter 获取 JSON 格式答案

【问题标题】：Kafka Connect HDFS Sink for JSON format using JsonConverterKafka Connect HDFS Sink 使用 JsonConverter 获取 JSON 格式
【发布时间】：2017-04-04 19:34:40
【问题描述】：

以 JSON 格式从 Kafka 生产/消费。使用以下属性以 JSON 格式保存到 HDFS：

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

制片人：

curl -X POST -H "Content-Type: application/vnd.kafka.json.v1+json" \
      --data '{"schema": {"type": "boolean", "optional": false, "name": "bool", "version": 2, "doc": "the documentation", "parameters": {"foo": "bar" }}, "payload": true }' "http://localhost:8082/topics/test_hdfs_json"

消费者：

./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-hdfs/quickstart-hdfs.properties

问题 1：

key.converter.schemas.enable=true

value.converter.schemas.enable=true

获取异常：

org.apache.kafka.connect.errors.DataException: JsonDeserializer with schemas.enable requires "schema" and "payload" fields and may not contain additional fields
    at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:332)

问题 2：

启用以上两个属性不会引发任何问题，但不会通过 hdfs 写入数据。

任何建议都将受到高度赞赏。

谢谢

【问题讨论】：

标签： apache-kafka confluent-platform apache-kafka-connect

【解决方案1】：

转换器指的是如何将数据从 Kafka 主题转换为由连接器解释并写入 HDFS。 HDFS 连接器仅支持开箱即用的 avro 或 parquet 写入 HDFS。您可以找到有关如何将格式扩展为 JSON here 的信息。如果您进行此类扩展，我鼓励您将其贡献给连接器的开源项目。

【讨论】：

感谢您的建议！
@dawsaw 你知道这样的扩展是否可以使用原生 kafka connect api 实现？
有一个 JsonConverter 已经随 Kafka 一起提供。我认为这里的问题特定于 HDFS 连接器的输出格式，这必然意味着扩展连接器，如果我正确理解了您的问题，则不使用 Connect 本身做任何事情。

【解决方案2】：

要写入HDFS的输入Json格式消息，请设置以下属性

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

【讨论】：

将检查 Akshat。感谢您的评论