什么是连接器？
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> 编码实现
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() { } } }