Flink1.11 Jdcb方式写mysql测试用例

 本demo是计算窗口内wordCount然后写入mysql

//数据通过jdbc方式sink到mysqlwindowCounts.addSink(JdbcSink.sink("replace into flink_test(words,nums) values(?,?)",new JdbcStatementBuilder<Tuple2<String, Integer>>() {@Overridepublic void accept(PreparedStatement ps, Tuple2<String, Integer> t) throws SQLException {ps.setString(1,t.f0);ps.setInt(2,t.f1);System.out.println("数据为"+t.f0+":"+t.f1);}},JdbcExecutionOptions.builder().withBatchSize(3)   //此处需注意，默认的batchSize是5000
//                        .withBatchIntervalMs(3).build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://localhost:3306/superset?serverTimezone=UTC").withDriverName("com.mysql.jdbc.Driver").withUsername("root").withPassword("123456").build())).setParallelism(1);//这个地方也需要注意，如果是并行度为2，// 如果有三条数据，则可能不对写入到mysql，因为一个batchSize也是按线程划的，要一个线程的batchSize达到3才行

我们通过JDBC方式sink到mysql，这里需要注意两个地方

1. 这里默认的batchSize是5000 ，如果不设置，可能会导致你的数据不会写入到msyql

JdbcExecutionOptions.builder().withBatchSize(3)   //此处需注意，默认的batchSize是5000
//                        .withBatchIntervalMs(3).build(),

源码如下 

/*** JDBC sink batch options.*/
@PublicEvolving
public class JdbcExecutionOptions implements Serializable {public static final int DEFAULT_MAX_RETRY_TIMES = 3;private static final int DEFAULT_INTERVAL_MILLIS = 0;public static final int DEFAULT_SIZE = 5000;private final long batchIntervalMs;private final int batchSize;private final int maxRetries;

2. sink的并行度设置

sink的batchSize跟并行度有关，当某个线程的batchSize达到3时数据才会写入，所以要看我们的业务需求，比如写入topN，那设置一个并行度就行

这里我是想有3条数据就写入mysql，所以当我在本地测试而且没有设置并行度时，数据总是不能正确写入