Flink批处理写入MySQL

前言：
主要是使用到JDBCOutputFormat类去实现写入的功能。前提要在数据库中创建好对应的表和字段哦
依赖

  
        
            org.apache.flink
            flink-java
            1.14.0
        
        
            org.apache.flink
            flink-streaming-java_2.12
            1.14.0
            provided
        
        
            org.apache.flink
            flink-clients_2.12
            1.14.0
        
        
            org.apache.flink
            flink-jdbc_2.12
            1.10.0
            provided
        
    

导包：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.types.Row;

java代码：

public class toMySQL {
    public static void main(String[] args) throws Exception {
        JDBCOutputFormat jdbcOutput = JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setDBUrl("jdbc:mysql://localhost:3306/test?user=root&password=123456")
                .setQuery("insert into  people (id,name) values (?,?) ")
                //设置为每有2条数据就提交一次，这里可以不要也行
                .setBatchInterval(2)
                .finish();

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource data = env.fromElements("1,denny", "2,jack");
		//由于写入数据库的数据必须使用Row对象传输，所以通过map将数据转换成Row对象
        MapOperator map = data.map(new MapFunction() {
            @Override
            public Row map(String s) throws Exception {
                String[] ss = s.split(",");
                Row row = new Row(2);
                row.setField(0, Integer.parseInt(ss[0]));
                row.setField(1, ss[1]);
                return row;
            }
        });
        
        map.output(jdbcOutput);
        env.execute();
    }
}