使用Kafka Debezium连接器导入MongoDB数据至S3格式异常求助

针对你遇到的问题，核心原因通常是MongoDB源连接器未正确配置转换器，导致数据以纯JSON字符串而非带Schema的结构化格式流入Kafka，进而使得S3 Sink无法生成带类型的Parquet文件。以下是具体排查和修复步骤：

1. 修正MongoDB Source Connector的转换器配置 #

确保源连接器使用与MySQL场景一致的AVRO转换器，并指向Schema Registry，而非默认的字符串转换器。示例配置片段如下：

name=mongodb-source-user-profiles
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
tasks.max=1
connection.uri=mongodb://your-mongo-host:27017
database=test_db
collection=user_profiles
topic.prefix=mongo-

# 关键：配置AVRO转换器，绑定Schema Registry
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://your-schema-registry:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://your-schema-registry:8081

# 确保输出带Schema的结构化数据（部分MongoDB连接器版本需设置）
output.format=schema

2. 验证Kafka主题中的消息格式 #

用Kafka控制台消费者确认主题内的消息是否为带Schema的AVRO数据，而非纯JSON字符串：

kafka-console-consumer.sh \
  --bootstrap-server your-msk-broker:9092 \
  --topic mongo-test_db.user_profiles \
  --from-beginning \
  --property print.value=true \
  --property schema.registry.url=http://your-schema-registry:8081 \
  --property value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

如果输出是结构化的键值对（而非单引号包裹的JSON字符串），说明源端配置已修复；若仍为纯JSON，需重新检查源连接器的转换器参数。

3. 调整S3 Sink Connector的配置 #

确保Sink连接器使用与源端一致的AVRO转换器，并指定Parquet格式：

name=s3-sink-mongo-profiles
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=1
topics=mongo-test_db.user_profiles
s3.bucket.name=your-target-bucket
s3.region=us-east-1
format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
storage.class=io.confluent.connect.s3.storage.S3Storage

# 保持与源端一致的转换器配置
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://your-schema-registry:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://your-schema-registry:8081

partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
flush.size=1000

4. 核对Schema Registry中的Schema兼容性 #

确认已注册的Schema与MongoDBuser_profiles集合的字段类型完全匹配：

• 用Schema Registry API查看最新Schema：

  curl http://your-schema-registry:8081/subjects/mongo-test_db.user_profiles-value/versions/latest
  

• 对比MongoDB文档的字段类型（如ObjectId需映射为AVRO的string，数值类型需匹配int/long/float等），避免因类型不匹配导致转换器降级为字符串输出。

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