flink: 通过Sink将数据写入MySQL

一、���赖添加

（）

                    mysql            mysql-connector-java            5.1.30                            org.apache.flink            flink-connector-jdbc_2.12            1.11.1        

二、两种方式写MySQL：

package cn.edu.tju.demo;import org.apache.flink.api.common.functions.*;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.connector.jdbc.JdbcConnectionOptions;import org.apache.flink.connector.jdbc.JdbcSink;import org.apache.flink.streaming.api.collector.selector.OutputSelector;import org.apache.flink.streaming.api.datastream.*;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.co.CoMapFunction;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import org.apache.flink.streaming.connectors.redis.RedisSink;import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;import org.apache.flink.util.Collector;import org.apache.http.HttpHost;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.apache.kafka.common.serialization.StringSerializer;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.Requests;import java.util.*;public class Test16 {    private static String MYSQL_SERVER = "139.198.0.216";    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment environment = StreamExecutionEnvironment                .getExecutionEnvironment();        DataStream mySource = environment.readTextFile("demo.txt");        String sql = "insert into userinfo(username,passwd) values  " +                "(?,?)";        JdbcConnectionOptions jdbcBuild = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()                .withDriverName("com.mysql.jdbc.Driver")                .withUrl("jdbc:mysql://xx.xx.xx.xx:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false")                .withUsername("root")                .withPassword("123456")                .build();        mySource.addSink(JdbcSink.sink(sql, (ps, str) -> {            ps.setString(1, str);            ps.setString(2, str);        }, jdbcBuild));        environment.execute("my job");    }}

package cn.edu.tju.demo;import org.apache.flink.api.common.functions.*;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.connector.jdbc.JdbcConnectionOptions;import org.apache.flink.connector.jdbc.JdbcSink;import org.apache.flink.streaming.api.collector.selector.OutputSelector;import org.apache.flink.streaming.api.datastream.*;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.co.CoMapFunction;import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;import org.apache.flink.streaming.api.functions.sink.SinkFunction;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import org.apache.flink.streaming.connectors.redis.RedisSink;import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;import org.apache.flink.util.Collector;import org.apache.http.HttpHost;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.apache.kafka.common.serialization.StringSerializer;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.Requests;import java.sql.Connection;import java.sql.DriverManager;import java.sql.PreparedStatement;import java.util.*;public class Test17 {    private static String MYSQL_SERVER = "xx.xx.xx.xx";    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment environment = StreamExecutionEnvironment                .getExecutionEnvironment();        DataStream mySource = environment.readTextFile("demo.txt");        mySource.addSink(new MyJdbcSinkFunction());        environment.execute("mysql");    }    public static class MyJdbcSinkFunction extends RichSinkFunction{        private Connection connection = null;        private PreparedStatement insert = null;        private PreparedStatement update = null;        @Override        public void open(Configuration parameters) throws Exception {            connection = DriverManager.getConnection("jdbc:mysql://xx.xx.xx.xx:3306/test" +                    "?useUnicode=true&characterEncoding=utf-8&useSSL=false", "root", "123456");            insert = connection.prepareStatement("insert into userinfo(username,passwd) values(?,?)");            System.out.println("open mysql");        }        @Override        public void close() throws Exception {            connection.close();        }        @Override        public void invoke(String value, Context context) throws Exception {            insert.setString(1, value);            insert.setString(2, value);            insert.execute();        }    }}

（）