开发Flink应用要求计算结果实时写入数据库的,一般业务写入TPS在600-800,如果生产同时跑十几个任务,数据库写入TPS接近一万,对数据库造成了较大压力,使用窗口的优化方向不可行:
1. 计算任务的key值较为分散(如用户,商户维度),小窗口(1分钟、5分钟)计算无法减少写入次数,大窗口(10分钟、1小时)实时性太差;
2. 无法保证上游流水100%有序准时到达,使用窗口计算容易漏算流水;
使用Flink应用对流水进行计算和统计,结果一般字段较少,每条计算结果的数据量不大,如果开启批量写入,可以降低和数据库之间的网络交换次数,也可提升数据库的数据写入效率;
配置此参数为true以后,jdbc驱动会在executeBatch时将SQL改写,将多条Insert语句合并为一条,效果如下:
改写前:
- INSERT INTO `t` (`a`) VALUES(10);
- INSERT INTO `t` (`a`) VALUES(11);
- INSERT INTO `t` (`a`) VALUES(12);
改写后:
INSERT INTO `t` (`a`) VALUES(10),(11),(12);
这个参数是用来节省数据库SQL解析和执行计划消耗的,开启useServerPrepStmts以后,服务器可以将已经解析的SQL反复使用,只在每次客户端提交新的写入请求时填入'?'占位符的参数;
值得注意的是,useServerPrepStmts开启后,客户端不再对SQL预编译,也就是说上面说的SQL改写也不会发生,所以在批量写入时,把此参数关闭,或保持默认配置;
关于rewriteBatchedStatements和useServerPrepStmts配合使用的性能测试,参考:
MySql PreparedStatement executeBatch过慢问题 - 灰信网(软件开发博客聚合)
虽然 useServerPrepStmts = true
能让服务端执行预处理语句,但默认情况下客户端每次执行完后会 close 预处理语句,并不会复用,这样预处理的效率甚至不如文本执行。所以建议开启 useServerPrepStmts = true
后同时配置 cachePrepStmts = true
,这会让客户端缓存预处理语句。
在配置 cachePrepStmts 后还需要注意 prepStmtCacheSqlLimit 配置(默认为 256
),该配置控制客户端缓存预处理语句的最大长度,超过该长度将不会被缓存。
在一些场景 SQL 的长度可能超过该配置,导致预处理 SQL 不能复用,建议根据应用 SQL 长度情况决定是否需要调大该值。
控制缓存的预处理语句数目(默认为 25
),如果应用需要预处理的 SQL 种类很多且希望复用预处理语句,可以调大该值。
jdbc:mysql://localhost:3306/mydb?charset=utf8mb4&useSSL=false&useConfigs=maxPerformance& cachePrepStmts=true&prepStmtCacheSqlLimit=8192&prepStmtCacheSize=1024&rewriteBatchedStatements=true&allowMultiQueries=true
原先SQL是这么写的:
- INSERT INTO
- rts_clear_branch_day_trans(stl_branch_id,
- stl_org_id,
- time_date,
- trans_count,
- trans_amount,
- mer_fee,
- acq_shr_amt)
- VALUES(?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE
- trans_count = ?,
- trans_amount = ?,
- mer_fee = ? ,
- acq_shr_amt = ?
改写后:
- INSERT INTO
- rts_clear_branch_day_trans(stl_branch_id,
- stl_org_id,
- time_date,
- trans_count,
- trans_amount,
- mer_fee,
- acq_shr_amt)
- VALUES(?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE
- trans_count = VALUES(`trans_count`),
- trans_amount = VALUES(`trans_amount`),
- mer_fee = VALUES(`mer_fee`),
- acq_shr_amt = VALUES(`acq_shr_amt`)
new JdbcExecutionOptions.Builder().withBatchSize(100).withBatchIntervalMs(1000).withMaxRetries(3).build(),