Flink批处理写入MySQL

Flink批处理写入MySQL

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

  <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.14.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.14.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_2.12</artifactId>
            <version>1.10.0</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

导包：

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<String> data = env.fromElements("1,denny", "2,jack");
		//由于写入数据库的数据必须使用Row对象传输，所以通过map将数据转换成Row对象
        MapOperator<String, Row> map = data.map(new MapFunction<String, Row>() {
            @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();
    }
}