Confluent HDFS Sink Connector：字符串转Parquet报Avro schema必须为record错误

问题根源 #

你遇到的"Avro schema must be a record"错误，核心原因是Confluent HDFS Connector默认期望处理带有结构化Avro Record Schema的数据，但你的源Kafka Topic里是无结构的纯文本字符串，没有符合要求的Schema结构，导致连接器无法解析并转换为Parquet格式。

解决方案 #

我们可以通过调整Kafka Connect的转换器配置，给纯文本字符串添加一个简单的Record Schema，让HDFS Connector能正确处理它。下面是具体的配置修改步骤：

1. 修改独立模式配置文件（connect-standalone.properties） #

找到文件中的key.converter和value.converter配置，调整为以下内容：

# 键转换器（如果你的键也是纯文本，就用这个；如果键是null或者不需要，也可以保持默认）
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=true

# 值转换器：用StringConverter处理纯文本，同时启用schema生成
value.converter=org.apache.kafka.connect.storage.StringConverter
value.converter.schemas.enable=true

这里启用schemas.enable=true后，StringConverter会自动给每个纯文本字符串生成一个包含单个string类型字段的Record Schema，刚好满足HDFS Connector对Record类型Schema的要求。

2. 调整HDFS Connector配置（quickstart-hdfs.properties） #

除了常规的HDFS路径、Topic等配置，需要添加/确认以下几个关键配置：

# 指定输出格式为Parquet
format.class=io.confluent.connect.hdfs.parquet.ParquetFormat

# 确保连接器能识别自动生成的schema
schema.compatibility=BACKWARD

# 可选：自定义字段名（默认自动生成的字段名是"string"，这里可以改成你想要的名字，比如"content"）
value.converter.schema.field.name=content

3. 可选：手动定义Avro Schema包装纯文本 #

如果你不想用自动生成的schema，也可以手动定义一个简单的Avro Record Schema来包装你的纯文本。比如创建一个text-schema.avsc文件：

{
  "type": "record",
  "name": "TextRecord",
  "namespace": "com.example",
  "fields": [
    {"name": "content", "type": "string"}
  ]
}

然后在connect-standalone.properties里修改value.converter为AvroConverter，并指定这个schema：

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=mock://localhost:8081  # 无Schema Registry时用mock地址
value.converter.value.schema.file=/path/to/text-schema.avsc
value.converter.schemas.enable=true

如果你有真实的Schema Registry服务，把schema.registry.url改成你的服务地址即可，无需使用mock。

4. 重启Connect并验证 #

修改完配置后，重新启动独立模式的Kafka Connect：

connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-hdfs/quickstart-hdfs.properties

之后检查HDFS目标路径下是否生成了Parquet文件，也可以用parquet-tools等工具查看文件内容是否和源Topic的纯文本一致。

常见坑点提醒 #

• 确保HDFS Connector版本和你的confluentinc/cp-kafka-connect:4.0.0镜像版本兼容，避免出现依赖冲突
• 如果Topic里存在空消息或非字符串格式的消息，建议添加errors.tolerance=all配置，避免任务直接崩溃
• 检查HDFS目标路径的权限，确保Connect进程拥有读写该路径的权限

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