storm学习篇（三）——集成MySQL

朱雀 2021-05-12 11:55 486阅读 0赞

**以统计单词代码为例子，用jdbcClient直接执行SQL**

添加依赖pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        
        <modelVersion>4.0.0</modelVersion>
        <artifactId>storm-jdbc</artifactId>
        <packaging>jar</packaging>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.storm</groupId>
                <artifactId>storm-core</artifactId>
                <version>1.1.1</version>
                
            </dependency>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.12</version>
                <scope>test</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.storm</groupId>
                <artifactId>storm-jdbc</artifactId>
                <version>1.0.1</version>
            </dependency>
    
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.31</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                        <archive>
                            <manifest>
                                <mainClass>cn.ljh.storm.helloworld.ExclamationTopology</mainClass>
                            </manifest>
                        </archive>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    
    
    </project>

创建spout：WordCountSpout.java

package com.storm.demo;
    
    import org.apache.storm.spout.SpoutOutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.topology.base.BaseRichSpout;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Values;
    import org.apache.storm.utils.Utils;
    
    import java.util.Map;
    import java.util.Random;
    
    /**
     * 路径：com.storm.demo
     * 类名：
     * 功能：《用一句话描述一下》
     * 备注：
     * 创建人：typ
     * 创建时间：2018/7/25 15:51
     * 修改人：
     * 修改备注：
     * 修改时间：
     */
    public class WordCountSpout extends BaseRichSpout {
    
        private SpoutOutputCollector collector;
    
        private static final String[] words = new String[]{"aaa","bbb","cad","adv","ed","eaf"};
    
        public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
            this.collector = spoutOutputCollector;
        }
        
        /**
         * 方法名：
         * 功能：把每一行数据发射出去
         * 描述：
         * 创建人：typ
         * 创建时间：2018/7/25 15:58
         * 修改人：
         * 修改描述：
         * 修改时间：
         */
        public void nextTuple() {
            Random random = new Random();
            String word = words[random.nextInt(words.length)];
            //发射出去
            this.collector.emit(new Values(word));
            System.out.println("emit==="+word);
    //        Utils.sleep(1000);
        }
    
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word"));
        }
    }

创建bolt：CountBolt.java

package com.storm.demo;
    
    import com.google.common.collect.Maps;
    import org.apache.storm.jdbc.common.Column;
    import org.apache.storm.jdbc.common.ConnectionProvider;
    import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
    import org.apache.storm.jdbc.common.JdbcClient;
    import org.apache.storm.task.OutputCollector;
    import org.apache.storm.task.TopologyContext;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.topology.base.BaseRichBolt;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    
    import java.sql.Types;
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    /**
     * 路径：com.storm.demo
     * 类名：
     * 功能：词频汇总Bolt
     * 备注：
     * 创建人：typ
     * 创建时间：2018/7/25 16:09
     * 修改人：
     * 修改备注：
     * 修改时间：
     */
    public class CountBolt extends BaseRichBolt {
    
        private OutputCollector collector;
        private JdbcClient jdbcClient;
        private ConnectionProvider connectionProvider;
    
        Map<String,Integer> map = new HashMap<String,Integer>();
    
        public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.collector = outputCollector;
            Map hikariConfigMap = Maps.newHashMap();
            hikariConfigMap.put("dataSourceClassName","com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
            hikariConfigMap.put("dataSource.url", "jdbc:mysql://localhost/mc_config");
            hikariConfigMap.put("dataSource.user","root");
            hikariConfigMap.put("dataSource.password","admin");
            connectionProvider = new HikariCPConnectionProvider(hikariConfigMap);
            //对数据库连接进行初始化
            connectionProvider.prepare();
            jdbcClient = new JdbcClient(connectionProvider,30);
        }
    
        /**
         * 方法名：
         * 功能：业务逻辑
         *     1.获取每个单词
         *     2.对所有单词进行汇总
         *     3.输出
         * 描述：
         * 创建人：typ
         * 创建时间：2018/7/25 16:17
         * 修改人：
         * 修改描述：
         * 修改时间：
         */
        public void execute(Tuple tuple) {
            String word = tuple.getStringByField("word");
            Integer count = map.get(word);
            if(count == null){
                count = 0;
            }
            count++;
            map.put(word,count);
            //查询该word是否存在
            List<Column> list = new ArrayList<Column>();
            //创建一列将值传入   列名  值  值的类型
            list.add(new Column("word",word, Types.VARCHAR));
            List<List<Column>> select = jdbcClient.select("select word from wordcount where word = ?",list);
            //计算出查询的条数
            Long n = select.stream().count();
            if(n>=1){
                //update
                jdbcClient.executeSql("update wordcount set word_count = "+map.get(word)+" where word = '"+word+"'");
            }else{
                //insert
                jdbcClient.executeSql("insert into wordcount values( '"+word+"',"+map.get(word)+")");
            }
    
        }
    
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            //后面jdbc insert bolt直接把这里的输出写Mysql里去了，所以这里的fileds的名字要跟mysql表的字段名字对应
            declarer.declare(new Fields("word","word_count"));
        }
    }

创建topology ：WordCountStormJdbcTopology.java

package com.storm.demo;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.topology.TopologyBuilder;
    
    /**
     * 路径：com.storm.demo
     * 类名：
     * 功能：《用一句话描述一下》
     * 备注：
     * 创建人：typ
     * 创建时间：2018/7/25 16:36
     * 修改人：
     * 修改备注：
     * 修改时间：
     */
    public class WordCountStormJdbcTopology {
    
        public static void main(String[] args) {
    
            TopologyBuilder builder = new TopologyBuilder();
            builder.setSpout("spout", new WordCountSpout());
            builder.setBolt("bolt", new CountBolt()).shuffleGrouping("spout");
    
            Config config = new Config();
            config.setNumWorkers(3);
    
            //本地模式
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("stormJdbcTopology", config, builder.createTopology());
    
    
    
        }
    
    }

运行结果：

![70][]

[70]: /images/20210511/9d6115e9ccb74aa08c2749f928fa09f9.png