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

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

//数据通过jdbc方式sink到mysql
        windowCounts.addSink(JdbcSink.sink("replace into flink_test(words,nums) values(?,?)",
                new JdbcStatementBuilder<Tuple2<String, Integer>>() {
                    @Override
                    public 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，所以当我在本地测试而且没有设置并行度时，数据总是不能正确写入