官方地址:

引入依赖

 <!--  -->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-java</artifactId>            <version>1.11.1</version>        </dependency>        <!--  -->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-streaming-java_2.12</artifactId>            <version>1.11.1</version><!--            <scope>provided</scope>-->        </dependency>        <!--  -->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-runtime-web_2.12</artifactId>            <version>1.11.1</version>            <scope>test</scope>        </dependency>        <!--  -->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-clients_2.12</artifactId>            <version>1.11.1</version>        </dependency>        <!--Flink JDBC  Sink-->        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-connector-jdbc_2.11</artifactId>            <version>1.11.2</version>        </dependency>       <!--mysql -->        <dependency>            <groupId>mysql</groupId>            <artifactId>mysql-connector-java</artifactId>            <version>8.0.20</version>        </dependency>        <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <optional>true</optional>        </dependency>

@Datapublic class Books {    private int id;    private String title;    private String author;    private double price;    private int qty;}

数据库名称：flink_jdbc建表语句：CREATE TABLE `books` (  `id` int(11) DEFAULT NULL,  `title` varchar(20) DEFAULT NULL,  `author` varchar(20) DEFAULT NULL,  `price` double(10,2) DEFAULT NULL,  `qty` int(11) DEFAULT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8

import com.bbx.flink.demo.entity.Books;import org.apache.flink.connector.jdbc.JdbcConnectionOptions;import org.apache.flink.connector.jdbc.JdbcSink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class JDBCDemo {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        Books books1 = new Books();        books1.setId(1);        books1.setAuthor("sss");        books1.setTitle("sss");        books1.setPrice(22.45);        books1.setQty(1);        env.fromElements(books1)                .addSink(JdbcSink.sink(                        "insert into books (id,title,author,price,qty)values(?,?,?,?,?) ",                        (ps, t) -> {                            ps.setInt(1, t.getId());                            ps.setString(2, t.getTitle());                            ps.setString(3, t.getAuthor());                            ps.setDouble(4, t.getPrice());                            ps.setInt(5, t.getQty());                        },                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()                                .withUrl("jdbc:mysql://192.168.10.236:3306/flink_jdbc?characterEncoding=UTF-8&serverTimezone=GMT%2B8")                                .withDriverName("com.mysql.cj.jdbc.Driver")                                .withUsername("root")                                .withPassword("123456")                                .build()                ));        env.execute();    }