Flink(kafka---＞mysql)

文章目录

maven pom表对应实体类Druid数据库连接池Mysql sink端Kafka source端

maven pom

    
        UTF-8
        1.11.2
        2.11
        2.11.12
    

    
        
            com.alibaba
            druid
            1.0.18
        

        
            org.apache.flink
            flink-walkthrough-common_${scala.binary.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-streaming-scala_${scala.binary.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-clients_${scala.binary.version}
            ${flink.version}
        

        
            com.alibaba
            fastjson
            1.2.47
        

        
            org.apache.flink
            flink-connector-kafka_2.11
            1.11.2
        

        
            mysql
            mysql-connector-java
            5.1.17

表对应实体类

import java.math.BigDecimal;
import java.util.Date;


public class Table {
    private String precinct_id;
    private String precinct_name;
    private String device_id;
    private String mete_code;
    private String mete_id;
    private BigDecimal report_value;
    private Date report_time;


    public String getPrecinct_id() {
        return precinct_id;
    }

    public void setPrecinct_id(String precinct_id) {
        this.precinct_id = precinct_id;
    }

    public String getPrecinct_name() {
        return precinct_name;
    }

    public void setPrecinct_name(String precinct_name) {
        this.precinct_name = precinct_name;
    }

    public String getDevice_id() {
        return device_id;
    }

    public void setDevice_id(String device_id) {
        this.device_id = device_id;
    }

    public String getMete_code() {
        return mete_code;
    }

    public void setMete_code(String mete_code) {
        this.mete_code = mete_code;
    }

    public String getMete_id() {
        return mete_id;
    }

    public void setMete_id(String mete_id) {
        this.mete_id = mete_id;
    }

    public BigDecimal getReport_value() {
        return report_value;
    }

    public void setReport_value(BigDecimal report_value) {
        this.report_value = report_value;
    }

    public Date getReport_time() {
        return report_time;
    }

    public void setReport_time(Date report_time) {
        this.report_time = report_time;
    }
}

Druid数据库连接池

import com.alibaba.druid.pool.DruidDataSource;
import java.sql.Connection;


public class DbUtils {
    private static DruidDataSource dataSource;

    public static Connection getConnection() throws Exception {
        dataSource = new DruidDataSource();
        dataSource.setDriverClassName("com.mysql.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://IP地址:端口号/库名?useUnicode=true&characterEncoding=utf8&characterSetResults=utf8");
        dataSource.setUsername("root");
        dataSource.setPassword("*****");
        //设置初始化连接数，最大连接数，最小闲置数
        dataSource.setInitialSize(10);
        dataSource.setMaxActive(50);
        dataSource.setMinIdle(5);
        //返回连接
        return  dataSource.getConnection();
    }
}

Mysql sink端

import java.sql.Timestamp;
import java.util.List;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;


public class MysqlJdbcSink extends RichSinkFunction> {
    private PreparedStatement insertPS;
    private Connection connection;

    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //获取数据库连接，准备写入数据库
        connection = DbUtils.getConnection();
        String insertSql = "replace into dc_current_voltage_test(precinct_id,precinct_name,device_id,mete_code,mete_id,report_value,report_time) values (?, ?, ?, ?, ?, ?, ?); ";
        insertPS = connection.prepareStatement(insertSql);
    }

    
    @Override
    public void close() throws Exception {
        super.close();
        //关闭并释放资源
        if (connection != null) {
            connection.close();
        }

        if (insertPS != null) {
            insertPS.close();
        }
    }

    
    @Override
    public void invoke(List tables, Context context) throws Exception {
        for (Table table : tables) {
            insertPS.setString(1, table.getPrecinct_id());
            insertPS.setString(2, table.getPrecinct_name());
            insertPS.setString(3, table.getDevice_id());
            insertPS.setString(4, table.getMete_code());
            insertPS.setString(5, table.getMete_id());
            insertPS.setBigDecimal(6, table.getReport_value());
            insertPS.setTimestamp(7, new Timestamp(table.getReport_time().getTime()));
            insertPS.addBatch();
        }

        //一次性写入
        int[] count = insertPS.executeBatch();
        System.out.println("成功写入Mysql数量：" + count.length);

    }
}
 
Kafka source端 
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import java.util.List;
import java.util.Properties;

public class KafkaSource {
    public static void main(String[] args) throws Exception {
        //构建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //kafka
        Properties props = new Properties();
        props.put("bootstrap.servers", "IP地址:端口");
        props.put("group.id", "kafka_mysql");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        DataStreamSource dataStreamSource = env.addSource(new FlinkKafkaConsumer(
                "topic01",
                new SimpleStringSchema(),
                props
        )).
                //单线程打印，控制台不乱序，不影响结果
                        setParallelism(1);

        //dataStreamSource.print();

        //从kafka里读取数据，转换成Table对象
        DataStream
 dataStream = dataStreamSource.map(new MapFunction() {
            @Override
            public Table map(String value) throws Exception {
                return JSONObject.parseObject(value, Table.class);
            }
        });
        //收集5秒钟的总数
        dataStream.timeWindowAll(Time.seconds(5L)).
                apply(new AllWindowFunction, TimeWindow>() {

                    @Override
                    public void apply(TimeWindow timeWindow, Iterable
 iterable, Collector> out) throws Exception {
                        List Tables = Lists.newArrayList(iterable);

                        if (Tables.size() > 0) {
                            System.out.println("5秒的总共收到的条数：" + Tables.size());
                            out.collect(Tables);
                        }

                    }
                })
                //sink 到数据库
                .addSink(new MysqlJdbcSink());
        //打印到控制台
        //.print();


        env.execute("kafka 消费任务开始");
    }
}

转载请注明：文章转载自 www.mshxw.com
本文地址：https://www.mshxw.com/it/755016.html

   上一篇  Python极速入门二（写第一个程序）
 
   下一篇  java 1-10进制的随意转换
  



Java相关栏目本月热门文章

  1【Linux驱动开发】设备树详解（二）设备树语法详解
  2别跟客户扯细节
  3Springboot+RabbitMQ+ACK机制(生产方确认(全局、局部)、消费方确认)、知识盲区
  4【Java】对象处理流（ObjectOutputStream和ObjectInputStream）
  5【分页】常见两种SpringBoot项目中分页技巧
  6一文带你搞懂OAuth2.0
  7我要写整个中文互联网界最牛逼的JVM系列教程 | 「JVM与Java体系架构」章节：虚拟机与Java虚拟机介绍
  8【Spring Cloud】新闻头条微服务项目：FreeMarker模板引擎实现文章静态页面生成
  9JavaSE - 封装、static成员和内部类
  10树莓派mjpg-streamer实现监控及拍照功能调试
  11用c++写一个蓝屏代码
  12从JDK8源码中看ArrayList和LinkedList的区别
  13idea 1、报错java: 找不到符号 符号: 变量 log 2、转换成Maven项目
  14在openwrt使用C语言增加ubus接口（包含C uci操作）
  15Spring 解决循环依赖
  16SpringMVC——基于MVC架构的Spring框架
  17Andy‘s First Dictionary C++ STL set应用
  18动态内存管理
  19我的创作纪念日
  20Docker自定义镜像-Dockerfile




热门相关搜索



路由器设置
木托盘
宝塔面板
儿童python教程
心情低落
朋友圈
vim
双一流学科
专升本
我的学校
日记学校
西点培训学校
汽修学校
情书
化妆学校
塔沟武校
异形模板
西南大学排名
最精辟人生短句
6步教你追回被骗的钱
南昌大学排名
清朝十二帝
北京印刷学院排名
北方工业大学排名
北京航空航天大学排名
首都经济贸易大学排名
中国传媒大学排名
首都师范大学排名
中国地质大学(北京)排名
北京信息科技大学排名
中央民族大学排名
北京舞蹈学院排名
北京电影学院排名
中国戏曲学院排名
河北政法职业学院排名
河北经贸大学排名
天津中德应用技术大学排名
天津医学高等专科学校排名
天津美术学院排名
天津音乐学院排名
天津工业大学排名
北京工业大学耿丹学院排名
北京警察学院排名
天津科技大学排名
北京邮电大学(宏福校区)排名
北京网络职业学院排名
北京大学医学部排名
河北科技大学排名
河北地质大学排名
河北体育学院排名






学习工具
代数计算器
三角函数
解析几何
立体几何


知识解答
教育知识
百科知识
生活知识
常识知识


写作必备
作文大全
作文素材
句子大全

实用范文


关于我们
关于我们
联系我们
网站地图


 
名师互学网交流群


名师互学网客服





名师互学网 版权所有 (c)2021-2022      ICP备案号：晋ICP备2021003244-6号
 




关于我们
文章归档
网站地图
联系我们
版权所有 (c)2021-2022 MSHXW.COM
ICP备案号：晋ICP备2021003244-6号