JDBC Sink Connector写入PostgreSQL的插入逻辑及优化方案咨询

问题1：JDBC Sink Connector写入PostgreSQL的运行逻辑与默认插入模式 #

你当前的默认配置对应的是逐行单条插入（方案1），运行逻辑如下：

• 每个Connector Task对应消费Kafka Topic的1个分区，拉取消息后会按默认batch.size=3000的阈值攒待写入的记录批次
• 但默认情况下，Connector生成的是单条INSERT语句，调用JDBC的batch接口攒批后，PostgreSQL官方JDBC驱动不会自动合并这些单条语句，还是会逐个发送给数据库执行，实际运行效果就是逐行插入。

问题2：批量合并插入（方案2）的开启方法 #

可以实现批量合并插入，只需要做以下配置调整：

必改配置 #

在JDBC连接URL中添加PostgreSQL驱动专属的批量重写参数，修改后的connection.url如下：

jdbc:postgresql://localhost:5432/postgres?reWriteBatchedInserts=true

这个参数会让JDBC驱动自动把多条单条INSERT语句，合并为一条多值的批量插入语句（格式为INSERT INTO 表名 (字段1,字段2) VALUES (值1,值2),(值3,值4)...），单次发送给数据库执行，可降低90%以上的数据库交互开销，写入性能会有几倍的提升。

可选优化配置 #

你可以根据实际消息大小调整攒批阈值，进一步放大批量写入的收益：

• 添加batch.size配置：设置单次批量写入的最大记录数，默认3000，可调整到5000~10000，不要设置过大避免单条SQL长度超出数据库限制
• 添加consumer.max.poll.records配置：设置消费者单次拉取的最大记录数，需要大于等于你设置的batch.size，保证能攒够批次
• 添加consumer.fetch.max.wait.ms=500配置：允许消费者最多等待500ms攒够批次再拉取，避免小批次频繁写入数据库

你当前设置的10个Topic分区对应10个tasks.max的配置是合理的，开启批量插入后性能提升会更明显。

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