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

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