借助Kafka Connect接收字节数组序列化Avro消息的配置问题

你现在用ByteArraySerializer把Avro消息序列化成字节数组发送到Kafka，要让HDFS Sink正确处理这些消息，得调整一下Connector的配置，我给你两个可行的方案：

方案1：直接解析原始Avro字节数组（无需Schema Registry） #

因为你发送的是纯Avro字节（没有Confluent序列化器添加的Schema ID前缀），所以需要让Connect直接传递字节数组，再让HDFS Sink用指定的Schema解析：

完整的HDFS Sink配置 #

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=csvtopic
hdfs.url=hdfs://10.15.167.119:8020
flush.size=3
locale=en-us
timezone=UTC
# 关键补充配置
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter.schemas.enable=false
format.class=io.confluent.connect.hdfs.avro.AvroFormat
# 替换成你的Avro Schema文件路径（Connect进程要能访问）
avro.schema.path=/path/to/your/avro-schema.avsc

配置说明 #

• value.converter=org.apache.kafka.connect.converters.ByteArrayConverter：让Connect原封不动地把消息的字节数组传递给HDFS Sink，不做额外的序列化/反序列化操作。
• value.converter.schemas.enable=false：关闭Connect的schema处理逻辑，因为我们要让AvroFormat自己用指定的Schema解析数据。
• format.class=io.confluent.connect.hdfs.avro.AvroFormat：指定Sink以Avro格式写入HDFS，它会自动用你指定的Schema解析传入的字节数组。
• avro.schema.path：一定要填你用来序列化消息的Avro Schema文件路径，不管是本地路径还是HDFS路径，只要Connect进程能访问到就行。

方案2：改用Confluent Avro序列化器（更推荐） #

如果能修改生产者的配置，我更建议用Confluent的Avro序列化器，这样消息里会带上Schema Registry的Schema ID，HDFS Sink可以自动从Schema Registry获取Schema，不用手动维护Schema文件：

修改后的生产者配置 #

key.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
# 替换成你的Schema Registry地址
schema.registry.url=http://your-schema-registry-host:8081

对应的HDFS Sink配置 #

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=csvtopic
hdfs.url=hdfs://10.15.167.119:8020
flush.size=3
locale=en-us
timezone=UTC
# 使用Avro转换器对接Schema Registry
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://your-schema-registry-host:8081
format.class=io.confluent.connect.hdfs.avro.AvroFormat

为什么推荐这个方案？ #

Schema可以在Schema Registry里集中管理，不用在多个地方维护Schema文件，而且后续Schema演进的时候兼容性更好，不用手动修改Sink的配置。

额外注意事项 #

• 如果用方案1，一定要确保Connect进程有权限访问你指定的Schema文件路径。
• 如果用方案2，要先确认Schema Registry已经注册了对应的Avro Schema，而且生产者能正常连接到Schema Registry。
• 不管用哪个方案，都要保证Connect进程有写入你指定HDFS路径的权限，不然会报权限错误。

内容的提问来源于stack exchange，提问作者Madhu