PyFlink CDC执行RDS到MySQL的Select Insert作业内存溢出如何解决

内存持续上涨的核心原因是SQL中使用了DISTINCT去重算子：该算子需要维护所有已出现的去重键状态用于判断重复，若未配置状态TTL，所有历史键会永久保留，即使使用RocksDB状态后端，高频访问的热数据也会驻留内存，叠加10张目标表的写入开销，很容易占满2GB TaskManager内存。另外你提供的SQL存在两处语法问题：SELECT子句末尾多了逗号、查询字段顺序和目标表字段顺序不匹配，也可能导致额外的异常开销。

• 优化去重逻辑，优先去掉不必要的DISTINCT：源表是MySQL CDC同步的表，本身带主键逻辑，CDC流本身不会有重复数据，大部分场景下不需要额外加DISTINCT。如果业务确实需要去重，给表环境配置状态TTL自动清理过期状态：

# 示例：设置去重状态7天自动过期，可根据业务调整时长
t_env.get_config().set("table.exec.state.ttl", "7d")

• 调整MySQL CDC连接器配置：将server-id配置为范围值（比如并行度为3的话配置为'1000-1003'），避免多并行度下读取冲突导致重复消费增加状态量；同时新增'debezium.snapshot.locking.mode' = 'none'配置，减少快照阶段的内存占用。
• 优化JDBC Sink配置，在目标表的WITH参数中新增批量写入配置，减少sink端缓存数据的内存占用：

'connector'= 'jdbc',
'url' = 'jdbc:mysql://name1:3306/name1',
'table-name' = 't1',
'username' = 'user123',
'password' = '123',
-- 新增以下两个配置
'sink.buffer-flush.max-rows' = '1000',
'sink.buffer-flush.interval' = '1s'

• 多表写入共享源读取：10张小表的写入全部加到同一个StatementSet中，避免重复读取源表，减少一倍的读取和计算开销。
• 开启RocksDB内存托管，限制RocksDB的内存使用上限，避免堆外内存溢出：

env.get_config().set("state.backend.rocksdb.memory.managed", "true")
# 调整托管内存占比，给RocksDB分配更多可用内存
env.get_config().set("taskmanager.memory.managed.fraction", "0.4")

内容的提问来源于stack exchange，提问作者gy_e-Aa