使用 JsonConverter 的 JSON 格式的 Kafka Connect HDFS Sink [英] Kafka Connect HDFS Sink for JSON format using JsonConverter

问题描述

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

Produce to/Consume from Kafka in JSON. Save to HDFS in JSON using below properties :

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 上写入任何数据.

Enabling above two properties is not throwing any issue, but no data are written over hdfs.

任何建议将不胜感激.

谢谢

推荐答案

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

The converter refers to how the data will be translated from the Kafka topic to be interpreted by the connector and written to HDFS. The HDFS connector only supports writing to HDFS in avro or parquet out of the box. You can find the information on how to extend the format to JSON here. If you make such an extension I encourage you to contribute it to the open source project for the connector.

这篇关于使用 JsonConverter 的 JSON 格式的 Kafka Connect HDFS Sink的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！