S3拼花地板格式的KafkaS3源连接器

Question

我有使用Protobuf制作的主题事件。我可以使用Parquet格式的S3 sink连接器将主题事件成功地汇到S3存储桶中。现在我的S3存储桶中有类型的对象。镶木地板和.key.parket。使用以下配置，所有这些都按预期工作：

{
    "name": "s3-sink",
    "config": {
      "name": "s3-sink",
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
      "keys.format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
      "value.converter.schema.registry.url": "https://my-schema-registry",
      "value.converter.basic.auth.credentials.source": "USER_INFO",
      "value.converter.basic.auth.user.info": "MY_SR_API_KEY:MY_SR_API_SECRET",
      "store.kafka.keys": true,
      "parquet.codec": "none",
      "partitioner.class": "io.confluent.connect.storage.partitioner.HourlyPartitioner",
      "locale": "en-US",
      "s3.bucket.name": "my-bucket-123",
      "s3.region": "eu-west-1",
      "time.interval": "HOURLY",
      "flush.size": 1000,
      "tasks.max": 1,
      "topics.regex": "test-topic.*",
      "confluent.license": "",
      "confluent.topic.bootstrap.servers": "my-bootstrap-server",
      "confluent.topic.replication.factor": 3,
      "confluent.license.topic.replication.factor": 1,
      "confluent.topic.security.protocol": "SASL_SSL",
      "confluent.topic.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"MY_API_KEY\" password=\"MY_API_SECRET\";",
      "confluent.topic.sasl.mechanism": "PLAIN",
      "confluent.topic.ssl.endpoint.identification.algorithm": "https"
    }
  }
}

现在，我想使用Protobuf将< code > my-bucket-123 (< code > parquet 格式)的键和值放回到Kafka主题中。为此，我使用以下配置通过汇合设置了一个新的S3源连接器(< code > Confluent Inc/Kafka-connect-S3-Source:1 . 4 . 5 ):

{
    "name": "s3-source",
    "config": {
      "name": "s3-source",
      "dest.kafka.bootstrap.servers": "my-bootstrap-server",
      "dest.topic.replication.factor": 1,
      "dest.kafka.security.protocol": "SASL_SSL",
      "dest.kafka.sasl.mechanism": "PLAIN",
      "dest.kafka.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"MY_API_KEY\" password=\"MY_API_SECRET\";",
      "tasks.max": 1,
      "connector.class": "io.confluent.connect.s3.source.S3SourceConnector",
      "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
      "confluent.license": "",
      "confluent.topic.bootstrap.servers": "my-bootstrap-server",
      "confluent.topic.replication.factor": 3,
      "confluent.license.topic.replication.factor": 1,
      "confluent.topic.security.protocol": "SASL_SSL",
      "confluent.topic.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"MY_API_KEY\" password=\"MY_API_SECRET\";",
      "confluent.topic.sasl.mechanism": "PLAIN",
      "confluent.topic.ssl.endpoint.identification.algorithm": "https",
      "transforms": "AddPrefix",
      "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
      "transforms.AddPrefix.regex": ".*",
      "transforms.AddPrefix.replacement": "copy_of_$0",
      "s3.region": "eu-west-1",
      "s3.bucket.name": "my-bucket-123"
    }
}

通过使用上述配置，我无法启动S3源连接器。如果我使用上述配置和命令验证配置：

curl -X PUT -d @config.json --header "content-Type:application/json" http://localhost:8083/connector-plugins/S3SourceConnector/config/validate

我在format.class属性中得到以下错误：

"errors":[
               "Invalid value io.confluent.connect.s3.format.parquet.ParquetFormat for configuration format.class: Class io.confluent.connect.s3.format.parquet.ParquetFormat could not be found.",
               "Invalid value null for configuration format.class: Class must extend: io.confluent.connect.cloud.storage.source.StorageObjectFormat"
            ]

我开始认为这个S3源代码连接器不支持＜code＞Parquet＜/code＞格式。我试着用JSON、AVRO和BYTE格式对其进行验证，所有这些都没有问题。

深入研究S3 Source连接器jar文件（1.4.5），我没有找到一个用于Par

Jar文件中的格式

有人知道这里发生了什么吗？有没有其他方法可以将数据从 S3 - Parquet 格式放回我的 Kafka 集群中？

谢谢！

Answer 1

从ConFluent的S3源连接器留档：

开箱即用，连接器支持从S3读取Avro和JSON格式的数据。除了带模式的记录，连接器还支持在文本文件中导入不带模式的普通JSON记录，每行一条记录。通常，连接器可以接受提供格式接口实现的任何格式。

因此，这意味着您应该能够添加/plugin实现< code>parquet格式，但它不是内置的

格式的源代码:

https://github.com/confluentinc/kafka-connect-storage-cloud/blob/master/kafka-connect-s3/src/main/java/io/confluent/connect/s3/format/parquet/ParquetFormat.java