Flink1.11 write MySQL test cases in jdcb mode

  Ben demo Is in the calculation window wordCount And then write mysql

// Data is passed through jdbc The way sink To 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(" The data is "+t.f0+":"+t.f1);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(3)   // Note here , default batchSize yes 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);// This place also needs attention , If the parallelism is 2,
        //  If there are three pieces of data , May not be written to mysql, Because a batchSize It is also marked by thread , Need a thread batchSize achieve 3 Talent 

We go through JDBC The way sink To mysql, Two things need to be noted here

1.  The default here is batchSize yes 5000 , If not set , It may cause your data not to be written to msyql

JdbcExecutionOptions.builder()
                        .withBatchSize(3)   // Note here , default batchSize yes 5000
//                        .withBatchIntervalMs(3)
                        .build(),

Source code is as follows  

/**
 * 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 Parallelism settings for

sink Of batchSize It is related to parallelism , When a thread batchSize achieve 3 Data will be written , So it depends on our business needs , Such as writing topN, Then set a parallelism

Here I want to have 3 Data is written mysql, So when I test locally and do not set the parallelism , Data is always not written correctly