你好,这个问题和flink无关,看你主键实现机制吧,如果是自增,那就是mysql级别自动实现的自增,跟flink搭不上关系的。
小昌同学 <ccc0606fight...@163.com> 于2023年5月31日周三 09:41写道: > > 老师,你好,再请教一下,连接数与并行度有关系的话,如果插入数据的MySQL是有主键的话,是不是连接数据也就是并行度只能为1啦呀,如果是多个并行度的话,可能会造成主键冲突; > 感谢各位老师的指导 > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > | > ---- 回复的原邮件 ---- > | 发件人 | lxk<lxk7...@163.com> | > | 发送日期 | 2023年5月30日 14:30 | > | 收件人 | <user-zh@flink.apache.org> | > | 主题 | Re:flink jdbcsink 连接数的问题 | > hi, > jdbc创建链接是在SimpleJdbcConnectionProvider这个类中实现的,至于真正创建链接,则是由DriverManager来处理。 > 关于连接数,则是跟你的并行度有关。 > > > > > > > > > > > > > > > > > > 在 2023-05-30 13:55:57,"小昌同学" <ccc0606fight...@163.com> 写道: > 各位老师,请教一下关于flink jdbcsink 连接数的问题; > 我使用代码如下:在以下代码中,我查看了一下源码,并没有找到sink到MySQL时关于连接数的设定,请问这一块关于连接数的设定我应该怎么写呀; > 谢谢各位老师的指导 > > | > outPutInfoStream.addSink(JdbcSink.sink( > "REPLACE into InputInfo (breakCode, breakName, breakDuration, > breakRule,breakPrimaryKey,breakStep,breakStepType,breakTime,breakSendTime,breakArgs) > values (?,?,?,?,?,?,?,?,?,?)", > (statement, InPutInfo) -> { > statement.setString(1,InPutInfo.getBreakCode()); > statement.setString(2,InPutInfo.getBreakName()); > statement.setLong(3,InPutInfo.getBreakDuration()); > statement.setString(4,InPutInfo.getBreakRule()); > statement.setString(5,InPutInfo.getBreakPrimaryKey()); > statement.setString(6, InPutInfo.getBreakStep()); > statement.setString(7, InPutInfo.getBreakStepType()); > statement.setString(8,InPutInfo.getBreakTime()); > statement.setString(9, DateUtil.format(new Date())); > statement.setString(10, String.valueOf(InPutInfo.getBreakArgs())); > }, > JdbcExecutionOptions.builder() > .withBatchSize(10) > .withBatchIntervalMs(200) > .withMaxRetries(5) > .build(), > new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() > .withUrl("jdbc:mysql://111/iap?useSSL=false&useUnicode=false&failOverReadOnly=false&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&autoReconnectForPools=true&serverTimezone=Asia/Shanghai&autoReconnect=true") > .withDriverName("com.mysql.jdbc.Driver") > .withUsername("111") > .withPassword("111") > .build() > )).name("sink-mysql"); > | > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > |