Storm+HBASE+MySQL 实时读取Kafka信息计算存储

淡淡的烟草味﹌ 2022-05-23 07:59 178阅读 0赞

### 本文是通过Storm将生产出来的数据进行实时的计算统计，整理出来之后将数据写到hbase和mysql数据中，并将结果展示在前端页面上，页面展示部分在下一篇说明 ###

#### 题目要求 ####

一、机组运行数据清洗规则  
1、运行数据日期不是当日数据  
2、运行数据风速 为空||=-902||风速在 3～12之外  
3、运行数据功率 为空||=-902||功率在 -0.5\*1500~2\*1500之外  
二、清洗数据后存储HBase  
1、正常数据 & 不合理数据 全部存入HBase中  
2、划分两个表（Normal/Abnormal）；Rowkey设计：年月日时分秒\_机组编号；列：Value（把数据写入一个列中）  
三、实时监控报警  
对于正常数据监控异常指标，并输出到MySQL中记录，Web显示报警信息。  
规则：每5S监控30S内发电机温度高于80度以上5次，报警（机组编号、报警时间、报警描述：过去30S内发电机温度高于80  
度以上出现：6/10（次））

Storm实时计算部分

*  通过在终端启动了一个消费者查看使用kafka生产出来的数据被消费者消费的格式如下，是以逗号分割的一条数据流。  
    ![这里写图片描述][70]
 *  首先定义一个SplitdataBolt()用来将数据字段进行分割，并且将下面的步骤需要的字段发送到下一个bolt中  
    ![这里写图片描述][70 1]  
    ![这里写图片描述][70 2]
 *  机组运行数据清洗规则  
    定义FliterBolt()用来进行数据的筛选。定义一个flag标识，当符合条件时，将flag标志为false，然后在发射数据的时候根据flag的值进行发射，同时标记了streamid，用于后面数据的分流。其中主要的不同还是streamid的不同，其他的字段都相同。  
    筛选条件  
    根据是否为当天日期，风速的值和功率值来进行筛选。当天日期只选取了2016/1/1的数据为当天数据。  
    ![这里写图片描述][70 3]  
    为了验证数据的分流是否准确，使用了两个printbolt用来打印两个分流的数据。分流数据的使用在shuffleGrouping中加上分流数据的id，就可以将正常和不正常的进行分类分别使用两个bolt进行输出。  
    ![这里写图片描述][70 4]  
    输出的数据如下  
    ![这里写图片描述][70 5]
 *  清洗数据后存储HBase  
    之前已经将数据的分流标记好id，只需要定义两个hbaseBolt分别进行配置和数据的接收，并将数据写入到hbase中  
     两个mapper和hbasebolt的设置  
    ![这里写图片描述][70 6]  
     输入到hbase接收分流数据  
    ![这里写图片描述][70 7]  
    输入到hbase结果  
    可以看到rowkey和最后的结果和要求相同。  
    正常数据  
    ![这里写图片描述][70 8]  
    异常数据  
    ![这里写图片描述][70 9]
 *  实时监控报警（使用滑动窗口进行计算）  
    使用滑动窗口进行温度超过80度的数据的总和统计  
    定义WindowMonitor继承BaseWindowedBolt类，通过execute方法将一个窗口内的数据进行统计。通过后面的if语句进行条件的判断，将温度大于80的风机编号进行计数，使用map来进行数据的存储  
    ![这里写图片描述][70 10]  
     数据的发送：当一个窗口统计结束后，将map中key的value>5的map进行数据的发送  
    ![这里写图片描述][70 11]  
     对接数据流  
    要定义窗口的长度和间隔，并且指定streamId

builder.setBolt("Filter", new FliterBolt(), 1)
     .shuffleGrouping("data-spilter");
    builder.setBolt("window",new WindowMonitor().withWindow(new BaseWindowedBolt.Duration(30, TimeUnit.SECONDS),new BaseWindowedBolt.Duration(5, TimeUnit.SECONDS)),1)
    .shuffleGrouping("Filter","Normal");

*  将统计数据输入到mysql数据库中  
    定义一个PersistentBolt类来进行jdbc的连接  
    Mysql数据库的配置

private static Map<String,Object> hikariConfigMap = new HashMap<String, Object>(){ { put("dataSourceClassName", "com.mysql.jdbc.jdbc2.optional.MysqlDataSource"); put("dataSource.url", "jdbc:mysql://172.17.11.183:3306/ExceptionMonitor"); put("dataSource.user","root"); put("dataSource.password", "root"); }};

创建getJdbcInsertBolt方法，用来向数据库中插入数据  
首先定义  
List<> schemaColumns用来存放数据库中的表的字段名  
创建如下两个对象使用withInsertQuery方法来进行数据库的插入

public static JdbcInsertBolt getJdbcInsertBolt(){
    
            List<Column> schemaColumns = Lists.newArrayList(new Column("fan_no", Types.VARCHAR),
                    new Column("call_time", Types.VARCHAR), new Column("call_count", Types.INTEGER));
            JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(schemaColumns);
            JdbcInsertBolt jdbcInsertBolt = new JdbcInsertBolt(connectionProvider,simpleJdbcMapper)
                    .withInsertQuery("insert into exception_output(fan_no,call_time,call_count) values(?,?,?)")
                    .withQueryTimeoutSecs(50);
            return jdbcInsertBolt;
            //使用tablename进行插入数据，需要指定表中的所有字段
            /*String tableName="ExceptionOutput";
            JdbcMapper simpleJdbcMapper = new SimpleJdbcMapper(tableName, connectionProvider);
            JdbcInsertBolt jdbcInsertBolt = new JdbcInsertBolt(connectionProvider, simpleJdbcMapper)
                                            .withTableName("ExceptionOutput")
                                            .withQueryTimeoutSecs(50);*/
            //使用schemaColumns，可以指定字段要插入的字段
        }

MySQL数据库展示  
![这里写图片描述][70 12]  
[Storm部分计算代码][Storm]  
[写入到mysql中][mysql]  
主要还是要把Storm的上下游机制以及各种组件的原理搞清楚，代码就好写了

[70]: /images/20220523/93f09c0f18144d0d8b67bee95c67e183.png
[70 1]: /images/20220523/8f5c592883c8448ea9436fc75acc48a1.png
[70 2]: /images/20220523/e398fb01e69a4769851aafd97964135c.png
[70 3]: /images/20220523/4490722694064045aabd8eff31628960.png
[70 4]: /images/20220523/f9f66edba6984e3fa41cef6e961de843.png
[70 5]: /images/20220523/1bf5121f9f814491860e32f43bda5753.png
[70 6]: /images/20220523/137031b1ed6846d3ac7156f7646211d3.png
[70 7]: /images/20220523/71fd5ab415574a80ae56eb00119a1fb5.png
[70 8]: /images/20220523/1e1a9d9619ec40899f77a29e96f27edc.png
[70 9]: /images/20220523/2e71658608b849cf8b4b5317667295e5.png
[70 10]: /images/20220523/fe556bf9af544fd9a005e03ccd8a2b63.png
[70 11]: /images/20220523/4ae398fea1e34544a2fe76fe0aa4a45b.png
[70 12]: /images/20220523/53f672f766024b97a2e6780ddf603e0d.png
[Storm]: https://github.com/random0815/shuang_0815/tree/master/StromProject/StormLast
[mysql]: https://github.com/random0815/shuang_0815/blob/master/StromProject/StormLast/PersistentBolt.java