Kafka Connect S3 Sink 连接器配置与对象存储落盘实践

--- title: Kafka Connect S3 Sink 连接器配置与对象存储落盘实践 keywords: Kafka Connect, S3 Sink, TimeBasedPartitioner, JsonFormat, flush.size, plugin.path, REST API description: 配置并部署 S3 Sink 连接器，将 Kafka 主题按时间分区写入对象存储并控制落盘粒度与滚动周期。 tags: - JsonFormat - Kafka - Kafka Connect - REST API - S3 - S3 Sink - TimeBasedPartitioner - flush.size - plugin.path - 数据流 categories: - 文章资讯 - 技术教程 --- Connect Worker 基本配置： ``` bootstrap.servers=localhost:9092 key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false plugin.path=/usr/share/java,/etc/kafka-connect/jars rest.port=8083 ``` S3 Sink 连接器配置（s3-sink.json）： ``` { "name": "s3-sink", "config": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "tasks.max": "2", "topics": "orders", "s3.bucket.name": "my-bucket", "s3.region": "us-east-1", "aws.access.key.id": "AKIA...", "aws.secret.access.key": "SECRET", "format.class": "io.confluent.connect.s3.format.json.JsonFormat", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "flush.size": "1000", "rotate.interval.ms": "600000", "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner", "path.format": "yyyy/MM/dd/HH", "timezone": "UTC", "locale": "en", "timestamp.extractor": "Record", "schema.compatibility": "NONE" } } ``` 通过 REST 创建并查看状态： ``` curl -X POST -H "Content-Type: application/json" --data @s3-sink.json http://localhost:8083/connectors curl http://localhost:8083/connectors/s3-sink/status ```