kafka s3连接器到minio，采用镶木地板格式

Question

我使用 docker compose 来启动 3 个服务：zookeeper、kafka broker 和 minio-connector

当我在 minio-connector 中使用以下配置从 kafka 消费并将 JSON 格式的记录转储到 minio 时，这三个服务可以成功启动和连接：

1. 启动命令：

<代码>root@e1d1294c6fe6：/opt/bitnami/kafka/bin#./connect单机版。sh/plugins/连接器。properties/plugins/s3 sink.properties

    bootstrap.servers=kafka:9092
    plugin.path=/plugins
    key.converter=org.apache.kafka.connect.storage.StringConverter
    value.converter=org.apache.kafka.connect.storage.StringConverter
    internal.key.converter=org.apache.kafka.connect.json.JsonConverter
    internal.value.converter=org.apache.kafka.connect.json.JsonConverter
    internal.key.converter.schemas.enable=false
    internal.value.converter.schemas.enable=false
    offset.storage.file.filename=/tmp/connect.offsets 

    name=s3-sink 
    connector.class=io.confluent.connect.s3.S3SinkConnector    
    tasks.max=1 
    topics=202208.minio.connector.test 
    s3.region=us-east-1   
    s3.bucket.name=minioUsr 
    s3.part.size=5242880 
    flush.size=1    
    store.url=https://minio.kube.url    
    storage.class=io.confluent.connect.s3.storage.S3Storage    
    format.class=io.confluent.connect.s3.format.json.JsonFormat   
    schema.compatibility=NONE 
    schema.generator.class=io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator    
    partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner 

现在，我希望连接器使用记录并以镶木地板格式转储到 minio。Kafka和Zookeeper服务保持不变。我修改了连接器属性和 s3-sink.properties，但连接器无法启动。

    bootstrap.servers=kafka:9092
    plugin.path=/plugins
    internal.key.converter=org.apache.kafka.connect.json.JsonConverter
    internal.value.converter=org.apache.kafka.connect.json.JsonConverter
    internal.key.converter.schemas.enable=false
    internal.value.converter.schemas.enable=false
    offset.storage.file.filename=/tmp/connect.offsets
    key.converter=io.confluent.connect.avro.AvroConverter
    value.converter=io.confluent.connect.avro.AvroConverter
    key.converter.schemas.enable=false
    value.converter.schemas.enable=true
    key.converter.schema.registry.url=https://registry...:1443
    value.converter.schema.registry.url=https://registry...:1443

    name=s3-sink
    connector.class=io.confluent.connect.s3.S3SinkConnector
    tasks.max=1
    topics=202208.minio.connector.test
    s3.region=us-east-1
    s3.bucket.name=minioUsr
    s3.part.size=5242880
    flush.size=1
    store.url=https://minio.kube.url
    storage.class=io.confluent.connect.s3.storage.S3Storage
    schema.generator.class=io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator
    partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
    schema.compatibility=NONE
    format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
    enhanced.avro.schema.support=true

我的问题是：以上配置，连接器异常启动失败

    [2022-08-29 15:21:10,610] ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectStandalone:130)
    org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.connect.avro.AvroConverter for configuration value.converter: Class
    io.confluent.connect.avro.AvroConverter could not be found.
            at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:728)
            at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474)
            at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:467)
            at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:108)
            at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:129)
            at org.apache.kafka.connect.runtime.WorkerConfig.<init>(WorkerConfig.java:385)
            at org.apache.kafka.connect.runtime.standalone.StandaloneConfig.<init>(StandaloneConfig.java:42)
            at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:81)

我通过下载 confluentinc-kafka-connect-s3-10.1.0 zip 文件并手动解压缩并将所有 jar 复制到 /plugins/lib 来安装连接器。我发现了以下与镶木地板相关的罐子：

    root@e1d1294c6fe6:/opt/bitnami/kafka/bin# ls -1 /plugins/lib/*parquet*
    /plugins/lib/parquet-avro-1.11.1.jar
    /plugins/lib/parquet-column-1.11.1.jar
    /plugins/lib/parquet-common-1.11.1.jar
    /plugins/lib/parquet-encoding-1.11.1.jar
    /plugins/lib/parquet-format-structures-1.11.1.jar
    /plugins/lib/parquet-hadoop-1.11.1.jar

安装中缺少什么？配置是否需要进一步更改？

Answer 1

我必须手动安装avro转换器使用conFluent-Hub

>

下载合流集线器客户端：https://docs.confluent.io/5.5.1/connect/managing/confluent-hub/client.html

使用汇流集线器安装avro转换器，如下所示：Kafka Connect汇流S3接收器连接器：io.confluent.Connect.avro类。找不到AvroConverter