kafka connect S3连接器内存分配答案

【问题标题】：kafka connect S3 connector memory allocationkafka connect S3连接器内存分配
【发布时间】：2021-02-17 02:21:06
【问题描述】：

我的 S3 连接器（如下配置）由于 java.lang.OutOfMemoryError 而失败。

据我从不同的帖子（例如here）了解到，内存分配大约是（s3 部分大小）*（#kafka 分区）*（分区程序创建的#partitions）。

我正在读取 60 个分区的主题，我的分区程序可以创建几十个分区，最小 s3.part.size 为 5 MB（由连接器强制执行，我会满意 5%）。

这是否意味着我不能使用小于 60 * 5 * 50 = 15000 MB 的堆大小？这只是为了这个连接器？

还有其他配置可以帮助我减少分配的内存吗？

连接器配置：

{
    "name": "test-processed-to-fake-dlk-6",
    "config": {
        "name": "test-processed-to-fake-dlk-6",
        "connector.class": "io.confluent.connect.s3.S3SinkConnector",
        "s3.region": "us-east-1",
        "topics.dir": "",
        "flush.size": "400",
        "schema.compatibility": "NONE",
        "tasks.max": "1",
        "topics": "raw-events-dev",
        "s3.part.size": "5242880",
        "timezone": "UTC",
        "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
        "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
        "schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
        "value.converter": "com.naturalint.kafka.connect.storage.SafeJsonConverter",
        "storage.class": "io.confluent.connect.s3.storage.S3Storage",
        "s3.bucket.name": "eventstestkinesis",
        "rotate.schedule.interval.ms": "60000"
    }
}

【问题讨论】：

如果你有 60 个分区，你会想要更多的任务。您还应该在集群中安装更多的连接器工作程序，这样一台机器就不会完成所有工作
我发现对于我的用例来说，减少分区数量就足够了。我喜欢在数据吞吐量大得多的情况下在多台机器之间拆分工作的想法，但这也需要在同一台机器上处理同一分区（分区器，而不是 kafka）的事件？
不确定我是否理解这个问题。每个任务都会拉出一个 Kafka 分区，该分区会构建一个多部分 S3 请求，将单个文件放置在连接器分区器位置...关于吞吐量，如果您一次只有一个任务拉取数据/上传数据，那么我你会惊讶于 OOM
据我了解：我有 1 个任务，从 60 个分区收集数据，将使用的事件分成几十个文件，每个文件至少 5M，以 60 * ~50 * 5 结尾为文件分配了约 15000 MB 的内存。我仍然不确定我是否正确理解了它，但无论如何减少 kafka 分区的数量似乎是合理的，而且它让事情变得稳定
没有亲自测试过，但我不认为它实际上一次消耗和分配所有60个分区/文件。

标签： apache-kafka apache-kafka-connect

【解决方案1】：

基于this memory allocation explanation 和this explanation on number of partitions optimization article。

我的主题吞吐量小于 1Mb/s，所以 60 个分区似乎是一个巨大的过度杀伤。

我将分区数量减少到 6 个，现在情况看起来很稳定。

【讨论】：