Debezium同步1GB PostgreSQL至Kafka后占100GB存储，如何优化？

针对初始加载后Kafka主题存储占用远超源数据库的问题，可通过以下几类配置优化缩减存储开销：

1. 切换紧凑的序列化格式 #

JSON作为文本格式存在大量冗余，改用二进制序列化格式可大幅降低单条消息体积：

• 使用Avro：配置value.converter=io.confluent.connect.avro.AvroConverter，搭配Schema Registry管理元数据，二进制格式的压缩率远高于JSON。
• 使用Protobuf：配置value.converter=io.confluent.connect.protobuf.ProtobufConverter，同样依赖Schema Registry，性能和压缩比表现优异。

2. 启用Kafka消息压缩 #

在Debezium连接器配置中开启生产者压缩，利用Kafka内置压缩算法减少传输和存储体积：

• 配置producer.compression.type=lz4：lz4兼顾压缩速度和压缩比，适合大部分场景；若追求更高压缩比可选用gzip，但会增加CPU开销。

3. 过滤冗余事件与字段 #

精简捕获的事件范围和数据内容，避免不必要的数据写入：

• 排除DDL变更事件：如果不需要跟踪表结构变化，设置exclude.schema.changes=true，减少schema变更产生的额外消息。
• 过滤不必要的操作类型：通过include.operations=c,u,d仅捕获创建、更新、删除操作，排除无需的操作（如r代表快照读，初始加载后无需重复快照可调整）。
• 字段级过滤：对大表使用field.exclude.list排除冗余字段（如大文本、二进制字段），或用field.include.list仅保留业务必需字段，示例：field.exclude.list=public.large_table.blob_column,public.log_table.raw_data。

4. 精简Debezium事件结构 #

Debezium默认事件包含before、after、source等多层包装结构，可通过转换去掉冗余部分：

• 启用unwrap转换：配置transforms=unwrap、transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState，仅保留变更后的after数据；若需保留操作类型等元数据，可添加transforms.unwrap.add.fields=op,ts_ms。
• 关闭消息内schema嵌入：设置include.schema=false，避免每条消息重复携带schema定义（使用Avro/Protobuf时无需此配置，元数据由Schema Registry管理）。

5. 优化初始快照配置 #

如果初始加载的全量数据包含大量非必要历史数据，可调整以下配置：

• 自定义快照查询：使用snapshot.select.statement.overrides为指定表设置快照查询，仅加载必要数据，示例：snapshot.select.statement.overrides=public.order_table=SELECT id, user_id, amount FROM public.order_table WHERE created_at > '2023-01-01'。
• 调整快照模式：若业务允许仅同步schema而非全量数据，设置snapshot.mode=schema_only（仅适用于无需初始全量同步的场景）。

6. Kafka主题层面优化 #

从Kafka主题存储策略入手，减少长期存储压力：

• 调整数据保留策略：设置主题的retention.ms（按时间保留）或retention.bytes（按大小保留），自动删除过期数据，示例：retention.ms=259200000（保留3天）。
• 优化分段大小：设置segment.bytes=1073741824（1GB），配合压缩算法提高存储效率，减少分段文件的数量和冗余。

内容的提问来源于stack exchange，提问作者Stavros Koureas