这可能是一个非常简单的问题,所以我会提前道歉。我正在为kafka主题添加一个s3接收器连接器,conf文件如下:
{
"schemas.enable": "false",
"name": "my-s3-sink",
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"errors.log.enable": "true",
"errors.log.include.messages": "true",
"topics": [
"my-topic-name"
],
"errors.deadletterqueue.context.headers.enable": "true",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"flush.size": "2000",
"rotate.schedule.interval.ms": "600000",
"s3.bucket.name": "my-bucket-name",
"s3.object.tagging": "true",
"s3.region": "region",
"s3.part.size": "5242880",
"aws.access.key.id": "****",
"aws.secret.access.key": "****",
"s3.ssea.name": "AES256",
"s3.compression.type": "gzip",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"topics.dir": "",
"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
"partition.duration.ms": "3600000",
"path.format": "YYYY/MM/dd/HH",
"locale": "en-GB",
"timezone": "UTC"
}
这将以以下格式输出消息 {topic_name}/YYYY/MM/DD/HH/{message}
,我希望钥匙就在那里 YYYY/MM/DD/HH/{message}
. 经过一些研究,我发现为了从键中删除主题名,我必须编写一个自定义分区器来扩展和覆盖 TimeBasedPartitioner
. (下面是一个例子https://github.com/confluentinc/kafka-connect-storage-cloud/issues/321)
我的问题是,我现在不知道在哪里为分区器编写实际的代码,它应该位于哪里?基于时间的分区器似乎链接到confluent拥有的某种注册表,但是定制的分区器会去哪里,我如何在连接器的conf文件中引用这些代码?
1条答案
按热度按时间7d7tgy0s1#
您在一个单独的项目中编写代码,将其编译成jar,然后将其放置在每个connect worker的类路径上。
那么你可以从
partitioner.class