Confluent S3 Source Connector 1.4.5读取Parquet文件报错解决方案问询

问题根因 #

1. 你使用的1.4.5版本Confluent官方S3 Source连接器未内置Parquet格式解析能力，该版本仅支持JSON、AVRO、BYTE三种格式的读取，所以Jar包中不存在对应Parquet格式的实现类，配置校验直接报错。
2. 你配置中填写的io.confluent.connect.s3.format.parquet.ParquetFormat是S3 Sink连接器的写入类，没有实现Source端要求的io.confluent.connect.cloud.storage.source.StorageObjectFormat接口，不符合Source连接器的格式类要求。

可用解决方案 #

方案一：升级S3 Source连接器版本（优先推荐） #

Confluent从S3 Source连接器2.0.0版本开始正式内置Parquet读取支持，直接将连接器版本升级到2.0及以上稳定版（例如confluentinc/kafka-connect-s3-source:2.4.0），修改对应格式配置即可正常运行：

{
  "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat"
}

升级后无需大幅调整原有配置，即可直接读取S3桶内的Parquet文件，配合Protobuf Converter可直接将数据以原Protobuf格式写回Kafka，同时支持读取.key.parquet文件恢复原始消息的Key。

方案二：自定义扩展Parquet格式实现 #

如果受环境限制无法升级连接器版本，可自行开发符合Source端要求的Parquet格式实现类，实现StorageObjectFormat接口的相关逻辑，将编译后的Jar包放入连接器的插件目录即可使用，该方案开发成本较高，仅适合有定制化需求的场景。

方案三：用大数据组件中转同步 #

如果仅需要一次性批量同步历史数据，可使用Spark、Flink等组件读取S3内的Parquet文件，关联Schema Registry中的原有Protobuf Schema做格式校验后，直接通过Kafka Producer以Protobuf格式写入目标Topic，无需调整现有连接器部署。

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