Flink-Connectors(连接器)（1）JDBC_flink-connector-jdbc

什么是连接器

预定义的源和接收器

​ Flink内置了一些基本数据源和接收器，这些数据源和接收器始终可用。该预定义的数据源包括文件、Mysql、RabbitMq、Kafka、ES

等，同时也支持数据输出到文件、Mysql、RabbitMq、Kafka、ES等。

​ 简单的说：flink连接器就是将某些数据源加载与数据输出做了封装（连接器），我们只要引入对应的连接器依赖，即可快速的完成对数据源的加载以及数据的输出。

​ 例如我们使用JDBC 连接器，即可快速的使用JDBC从数据库中加载数据源并支持数据通过JDBC 输出到我们的数据库。

使用JDBC连接器 快速输出到Mysql

必要依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.12</artifactId>
    <version>1.12.2</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>
1
2
3
4
5
6
7
8
9
10

编码实现

package com.leilei;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/13 21:07
 * @desc flink jdbc 连接器
 */
public class FlinkConnector1_JDBC {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        DataStreamSource<VehicleAlarm> source = env.addSource(new MySource());
        String sql = "insert into vehicle_alarm_202103 (`id`,`license_plate`,`plate_color`,`device_time`,`zone`) " +
                "values(?,?,?,?,?)";
        JdbcConnectionOptions jdbcBuild = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName("com.mysql.jdbc.Driver")
                .withUrl("jdbc:mysql://xxxx:3306/alarm-sc?useUnicode=true&characterEncoding=utf-8&useSSL=false")
                .withUsername("root")
                .withPassword("root")
                .build();
        /**
         * 使用JDBC连接器 将数据发送到Mysql
         */
        source.addSink(JdbcSink.sink(sql, (ps, vehicle) -> {
            ps.setString(1, vehicle.getId());
            ps.setString(2, vehicle.getLicensePlate());
            ps.setString(3, vehicle.getPlateColor());
            ps.setLong(4, vehicle.getDeviceTime());
            ps.setString(5, vehicle.getZone());
        }, jdbcBuild));
        try {
            env.execute("jdbc-connector");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class VehicleAlarm {
        private String id;
        private String licensePlate;
        private String plateColor;
        private Long deviceTime;
        private String zone;
    }

    /**
     * 自定义数据源
     */
    public static class MySource extends RichSourceFunction<VehicleAlarm> {
     
        @Override
        public void run(SourceContext<VehicleAlarm> ctx) throws Exception {
                long id = System.currentTimeMillis()/1000;
                VehicleAlarm vehicleAlarm = new VehicleAlarm(String.valueOf(id), "川A" + id,
                        "紫", System.currentTimeMillis(), "sc");
                ctx.collect(vehicleAlarm);
                Thread.sleep(10000);
            }

        
        @Override
        public void cancel() {

        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82