Flink-Connectors(连接器)（1）JDBC

时间：2022-09-25 16:30:00 5023521000连接器 643330100连接器 430300001连接器 9060480505连接器 342300003连接器 19200030220连接器

什么是 连接器？

源和接收器的预定义

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

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

简单的说：flink连接器是一些数据源加载与数据输出包装(连接器)后，只要引入相应的连接器依赖，就能快速完成数据源的加载和数据的输出。

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

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

必要依赖

<dependency>     <groupId>org.apache.flinkgroupId>     <artifactId>flink-connector-jdbc_2.12artifactId>     <version>1.12.2version> dependency> <dependency>     <groupId>mysqlgroupId>     <artifactId>mysql-connector-javaartifactId>     <version>5.1.47version> 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() { 
        

        }
    }

}

锐单商城拥有海量元器件数据手册、IC替代型号，打造电子元器件IC百科大全！

Flink-Connectors(连接器)（1）JDBC

什么是连接器？

相关文章