Flink customizes the application of sink side sinkfunction

The scene that

 from kafka It reads data , Processed write to mysql Inside . stay flink Inside is from source To sink The process of , So this article explains ,mysqlsink Single data insertion and batch data insertion .

1、 Let's start with SinkFunction Inheritance relationship of

adopt API We can see sinkFunction There are many implementation classes for interfaces . For example, common RichSinkFunction.

2、mysql Single data insertion

public class SinkToMysql extends RichSinkFunction<Tuple2<String, Integer>> {
    
public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
    
        conn = getConnection();
        String sql = "insert into dm_stu(stuname, stuaddr) values(?,?);";
        ps = this.conn.prepareStatement(sql);
        ps.setString(1, value.f0);
        ps.setString(2, value.f0 + "1");
// ps.addBatch();
        int resultBatch = ps.executeUpdate();
        System.out.println(" Successfully inserted data  " + resultBatch + " That's ok ");
        this.close();
    }
}	

The above class implements RichSikFunction class , rewrite invoke Method , This method is used to perform interaction with the database , insert data . This method is executed once there is a piece of data upstream . The actual development efficiency is low .

3、MySQL Batch insert

When the program arrives sink When , We can accumulate the upstream data for a period of time , Send together in batches to sink Downstream of .

public class SinkToMysqlBatch implements SinkFunction<ArrayList<String>> {
    
    PreparedStatement ps;
    Connection conn;

    @Override
    public void invoke(ArrayList<String> value, Context context) throws Exception {
    
        conn = SinkToMysql.getConnection();
        conn.setAutoCommit(false);
        String sql = "insert into dm_stu(stuname, stuaddr) values(?,?);";
        ps = this.conn.prepareStatement(sql);
        for (String s: value) {
    
            ps.setString(1, s);
            ps.setString(2,s+1);
            ps.addBatch();
        }
        int[] resultBatch = ps.executeBatch();
        System.out.println(" Insert data into the current window ：" + resultBatch.length+ "  That's ok ");
        conn.commit();
        try {
    
            if (ps != null) {
    
                ps.close();
            }
            if (conn != null) {
    
                conn.close();
            }
        }catch (Exception e) {
    
            e.printStackTrace();
        }
    }

}

The above batch operation , The custom class implements SinkFunction Interface , Rewrote invoke Method . Generics are collection types , Indicates that the data passed from the upstream is a collection . We need to be in invoke Method to traverse . One batch Write to mysql.

The problem encountered in actual development is

In this code ,addSink( Customized in mysql Batch classes inherit RichSinkFunction class , But there will be an error here, saying that it must be realized SinkFunction Interface ). There are questions here .