Storm编程之wordcount（kafka--》Jstorm--》redis）

本文是笔者这周做的一个小小的尝试。

设计的软件比较多，大家可以一一在本机安装一下，我的电脑是mac pro，基本安装起来和Linux基本一致，比较简单

基本都是下载 解压包 拷贝到安装目录重命名 然后启动服务

需要安装的基本有 JDK1.7，IDEA，kafka，Jstorm，redis都是单机

下面直接给出项目的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>

    <groupId>KafkaStormTest</groupId>
    <artifactId>zfs_try</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>0.9.2-incubating</version>
            <scope>compile</scope>
            <exclusions>
                <exclusion>
                    <groupId>ch.qos.logback</groupId>
                    <artifactId>logback-classic</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>log4j-over-slf4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-kafka</artifactId>
            <version>0.9.2-incubating</version>
            <scope>compile</scope>
        </dependency>

        <!--<dependency>-->
            <!--<groupId>org.apache.kafka</groupId>-->
            <!--<artifactId>kafka_2.9.2</artifactId>-->
            <!--<version>0.8.1.1</version>-->
            <!--<exclusions>-->
                <!--<exclusion>-->
                    <!--<groupId>org.apache.zookeeper</groupId>-->
                    <!--<artifactId>zookeeper</artifactId>-->
                <!--</exclusion>-->
                <!--<exclusion>-->
                    <!--<groupId>log4j</groupId>-->
                    <!--<artifactId>log4j</artifactId>-->
                <!--</exclusion>-->
            <!--</exclusions>-->
        <!--</dependency>-->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.15</version>
        </dependency>
        <!--<dependency>-->
            <!--<groupId>org.apache.storm</groupId>-->
            <!--<artifactId>storm-core</artifactId>-->
            <!--<version>1.0.2</version>-->
            <!--<exclusions>-->
                <!--<exclusion>-->
                    <!--<groupId>org.apache.logging.log4j</groupId>-->
                    <!--<artifactId>log4j-slf4j-impl</artifactId>-->
                <!--</exclusion>-->
                <!--<exclusion>-->
                    <!--<groupId>org.slf4j</groupId>-->
                    <!--<artifactId>log4j-over-slf4j</artifactId>-->
                <!--</exclusion>-->
            <!--</exclusions>-->
        <!--</dependency>-->
        <!--<dependency>-->
            <!--<groupId>org.apache.storm</groupId>-->
            <!--<artifactId>storm-kafka</artifactId>-->
            <!--<version>1.0.2</version>-->
        <!--</dependency>-->
        <dependency>
            <groupId>commons-logging</groupId>
            <artifactId>commons-logging</artifactId>
            <version>1.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.10</artifactId>
            <version>0.10.1.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.7.2</version>
        </dependency>
    </dependencies>
</project>           

注意我本地安装的阿里的Jstorm

不是storm，如果你用storm的话请用0.9.*版本的，最新的版本我测试过貌似cluster这个类初始化有问题

LocalCluster cluster = new LocalCluster();      

我们先来看看main函数类的代码：其中主要配置spoutconf和初始化kafkaspout，如果看不懂可以直接看看官网的资料

https://github.com/apache/storm/blob/1.0.x-branch/external/storm-kafka/README.md

然后我为这个topology定义了两个bolt分别用于split word和wordcount 在wordcount中实现了数据load into redis

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;

import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.tuple.Fields;
import kafka.common.AuthorizationException;
import storm.kafka.*;
import storm.kafka.bolt.KafkaBolt;

import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
/**
 * Created by mac on 2016/12/5.
 */
public class KafkaReader {

    public static void main(String[] args) {
        if (args.length != 2) {
            System.err.println("Usage: inputPaht timeOffset");
            System.err.println("such as : java -jar WordCount.jar D://input/ 2");
            System.exit(2);
        }

        String zks = "127.0.0.1:2181";//,h2:2181,h3:2181
        String topic = "TEST-TOPIC";
        String zkRoot = "/storm"; // default zookeeper root configuration for storm
        String id = "word";

        BrokerHosts brokerHosts = new ZkHosts(zks);
        SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id);
        spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
        spoutConf.useStartOffsetTimeIfOffsetOutOfRange=true;
        spoutConf.forceFromStart= false;//.forceFromStart
        spoutConf.zkServers = Arrays.asList(new String[] {"127.0.0.1"});//"h1", "h2", "h3"
        spoutConf.zkPort = 2181;
        KafkaSpout kafkaSpout=new KafkaSpout(spoutConf);

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("kafka-reader",  kafkaSpout,1); // Kafka我们创建了一个5分区的Topic，这里并行度设置为5


        //builder.setSpout("word-reader", new WordReader());
        // builder.setBolt("word-spilter", new WordSpliter()).shuffleGrouping("word-reader");
        builder.setBolt("word-spilter", new WordSpliter()).shuffleGrouping("kafka-reader");
        builder.setBolt("word-counter", new WordCounter2()).shuffleGrouping("word-spilter");
        String inputPath = args[0];
        String timeOffset = args[1];
        Config conf = new Config();
        conf.put("INPUT_PATH", inputPath);
        conf.put("TIME_OFFSET", timeOffset);
        conf.setDebug(false);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("WordCount", conf, builder.createTopology());
    }

}
           

下面看下wordsplit类定义：

/**
 * Created by mac on 2016/12/5.
 */
import org.apache.commons.lang.StringUtils;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class WordSpliter extends BaseBasicBolt {

    private static final long serialVersionUID = -5653803832498574866L;


    public void execute(Tuple input, BasicOutputCollector collector) {
        String line = input.getString(0);
        String[] words = line.split(" ");
        for (String word : words) {
            word = word.trim();
            if (StringUtils.isNotBlank(word)) {
                word = word.toLowerCase();
                collector.emit(new Values(word));
            }
        }
    }


    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));

    }

}           

wordcount类的定义：

/**
 * Created by mac on 2016/12/5.
 */
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

public class WordCounter2 extends BaseBasicBolt {
    private static final long serialVersionUID = 5683648523524179434L;
    private HashMap<String, Integer> counters = new HashMap<String, Integer>();
    private volatile boolean edit = false;


    @SuppressWarnings("rawtypes")
    public void prepare(Map stormConf, TopologyContext context) {
        final long timeOffset = Long.parseLong(stormConf.get("TIME_OFFSET").toString());
        new Thread(new Runnable() {

            public void run() {
                while (true) {
                    if (edit) {
                        for (Entry<String, Integer> entry : counters.entrySet()) {
                            System.out.println(entry.getKey() + " : " + entry.getValue());
                        }
                        System.out.println("WordCounter---------------------------------------");
                                edit = false;
                    }
                    try {
                        Thread.sleep(timeOffset * 1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }

//load into redis
    public void execute(Tuple input, BasicOutputCollector collector) {
        String str = input.getString(0);

        RedisMethod jedistest = new RedisMethod();
        jedistest.setup();
        System.out.println("Connection to server sucessfully");



        if (!counters.containsKey(str)) {
            counters.put(str, 1);
        } else {
            Integer c = counters.get(str) + 1;
            counters.put(str, c);
        }
        jedistest.jedis.zincrby("storm2redis",1,str);


        edit = true;
        System.out.println("WordCounter+++++++++++++++++++++++++++++++++++++++++++");
    }


    public void declareOutputFields(OutputFieldsDeclarer declarer) {
//        declarer.declare(new Fields("key"));
//        declarer.declare(new Fields("value"));

    }
}           

我这里将相关redis的用法放到一个类里面是为了方便以后自己学习使用：

import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * Created by mac on 2016/12/6.
 */




public  class RedisMethod {

    public static void main(String[] args) {
        RedisMethod jedistest = new RedisMethod();
        jedistest.setup();
        System.out.println("Connection to server sucessfully");
        //查看服务是否运行

        jedistest.testzadd();

        //      Map<double, String> map = new HashMap<double, String>();
//
//        map.put(12, "hadoop");
//        map.put(11, "hbase");
//        map.put(13, "storm");
//        jedis.zadd("user",map);


    }
    public Jedis jedis;


    public  void setup() {
        //连接redis服务器，127.0.0.1:6379
        jedis = new Jedis("127.0.0.1", 6379);
        //权限认证
        jedis.auth("123456");
        System.out.println("Server is running: "+jedis.ping());

    }

    /**
     * redis存储字符串
     */
    public void testzadd() {
        jedis.zadd("zfs1",2,"hadoop");
        jedis.zadd("zfs1",1,"storm");
        jedis.zadd("zfs1",3,"hive");

        Map<String,Double> map = new HashMap<String,Double>();
        map.put("hadoop",12.0);
        map.put("hbase",11.0);
        map.put("storm",13.0);
        map.put("hadoop1",12.0);
        map.put("hbase1",11.0);
        map.put("storm1",13.0);
        jedis.zadd("mapdf",map);
    }
    public void testString() {
        //-----添加数据----------
        jedis.set("name","xinxin");//向key-->name中放入了value-->xinxin
        System.out.println(jedis.get("name"));//执行结果：xinxin

        jedis.append("name", " is my lover"); //拼接
        System.out.println(jedis.get("name"));

        jedis.del("name");  //删除某个键
        System.out.println(jedis.get("name"));
        //设置多个键值对
        jedis.mset("name","liuling","age","23","qq","476777XXX");
        jedis.incr("age"); //进行加1操作
        System.out.println(jedis.get("name") + "-" + jedis.get("age") + "-" + jedis.get("qq"));

    }

    /**
     * redis操作Map
     */

    public void testMap() {
        //-----添加数据----------
        Map<String, String> map = new HashMap<String, String>();
        map.put("name", "xinxin");
        map.put("age", "22");
        map.put("qq", "123456");
        jedis.hmset("user",map);
        //取出user中的name，执行结果:[minxr]-->注意结果是一个泛型的List
        //第一个参数是存入redis中map对象的key，后面跟的是放入map中的对象的key，后面的key可以跟多个，是可变参数
        List<String> rsmap = jedis.hmget("user", "name", "age", "qq");
        System.out.println(rsmap);

        //删除map中的某个键值
        jedis.hdel("user","age");
        System.out.println(jedis.hmget("user", "age")); //因为删除了，所以返回的是null
        System.out.println(jedis.hlen("user")); //返回key为user的键中存放的值的个数2
        System.out.println(jedis.exists("user"));//是否存在key为user的记录 返回true
        System.out.println(jedis.hkeys("user"));//返回map对象中的所有key
        System.out.println(jedis.hvals("user"));//返回map对象中的所有value

        Iterator<String> iter=jedis.hkeys("user").iterator();
        while (iter.hasNext()){
            String key = iter.next();
            System.out.println(key+":"+jedis.hmget("user",key));
        }
        jedis.hdel("user");
    }

    /**
     * jedis操作List
     */

    public void testList(){
        //开始前，先移除所有的内容
        jedis.del("java framework");
        System.out.println(jedis.lrange("java framework",0,-1));
        //先向key java framework中存放三条数据
        jedis.lpush("java framework","spring");
        jedis.lpush("java framework","struts");
        jedis.lpush("java framework","hibernate");
        //再取出所有数据jedis.lrange是按范围取出，
        // 第一个是key，第二个是起始位置，第三个是结束位置，jedis.llen获取长度 -1表示取得所有
        System.out.println(jedis.lrange("java framework",0,-1));

        jedis.del("java framework");
        jedis.rpush("java framework","spring");
        jedis.rpush("java framework","struts");
        jedis.rpush("java framework","hibernate");
        System.out.println(jedis.lrange("java framework",0,-1));
    }

    /**
     * jedis操作Set
     */

    public void testSet(){
        //添加
        jedis.del("user");
        jedis.sadd("user","liuling");
        jedis.sadd("user","xinxin");
        jedis.sadd("user","ling");
        jedis.sadd("user","zhangxinxin");
        jedis.sadd("user","who");
        //移除noname
        jedis.srem("user","who");
        System.out.println(jedis.smembers("user"));//获取所有加入的value
        System.out.println(jedis.sismember("user", "who"));//判断 who 是否是user集合的元素
        System.out.println(jedis.srandmember("user"));
        System.out.println(jedis.scard("user"));//返回集合的元素个数
    }


    public void test() throws InterruptedException {
        //jedis 排序
        //注意，此处的rpush和lpush是List的操作。是一个双向链表（但从表现来看的）
        jedis.del("a");//先清除数据，再加入数据进行测试
        jedis.rpush("a", "1");
        jedis.lpush("a","6");
        jedis.lpush("a","3");
        jedis.lpush("a","9");
        System.out.println(jedis.lrange("a",0,-1));// [9, 3, 6, 1]
        System.out.println(jedis.sort("a")); //[1, 3, 6, 9]  //输入排序后结果
        System.out.println(jedis.lrange("a",0,-1));
    }



}           

好，到这里已经已经贴了全部代码了。

先来看看结果

首先进入到kafka的客户端生产数据：

程序终端显示如下：

下面我们看看redis里面数值：

验证成功，下面来把代码打成jar包，因为这里我们用到了jedis的jar包，我们可以将这个jar先copy到Jstorm Lib目录下，或者直接打包时打到包内，将我们打的jar传到Jstorm Lib目录下，运行

bin/jstorm jar /Users/mac/jstorm-2.1.1/lib/KafkaStormTest.jar KafkaReader Users/mac/input 2 即可，后面的两位是参数，是我测试本地文件夹为spout的时用的，你可以改下代码不用可以去掉。