kafka+storm+hbase如何实现计算WordCount

"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">

  4.0.0

  com

  kafkaSpout

  0.0.1-SNAPSHOT

            org.apache.storm

            storm-core

            0.9.3

            org.apache.storm

            storm-kafka

            0.9.3

            org.apache.kafka

            kafka_2.10

            0.8.1.1

                    org.apache.zookeeper

                    zookeeper

                    log4j

                    log4j

            org.apache.hbase

            hbase-client

            0.99.2

                    org.slf4j

                    slf4j-log4j12

                    org.apache.zookeeper

                    zookeeper

         com.google.protobuf

         protobuf-java

         2.5.0

            org.apache.curator

            curator-framework

            2.5.0

                    log4j

                    log4j

                    org.slf4j

                    slf4j-log4j12

            jdk.tools

            jdk.tools

            1.7

            system

            C:\Program Files\Java\jdk1.7.0_51\lib\tools.jar

            central

            http://repo1.maven.org/maven2/

                false

                true

            clojars

            https://clojars.org/repo/

                true

                true

            scala-tools

            http://scala-tools.org/repo-releases

                true

                true

            conjars

            http://conjars.org/repo/

                true

                true

                org.apache.maven.plugins

                maven-compiler-plugin

                3.1

                    1.6

                    1.6

                    UTF-8

                    true

                    true

                maven-assembly-plugin

                        jar-with-dependencies

                        make-assembly

                        package

                            single

package com.kafka.spout;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

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 LevelSplit extends BaseBasicBolt {

    public void execute(Tuple tuple, BasicOutputCollector collector) {

        String words = tuple.getString(0).toString();//the cow jumped over the moon

        String []va=words.split(" ");

        for(String word : va)

        {

            collector.emit(new Values(word));

        }

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        declarer.declare(new Fields("word"));

    }

}

package com.kafka.spout;

import java.util.HashMap;

import java.util.Map;

import java.util.Map.Entry;

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 LevelCount extends BaseBasicBolt {

    Map counts = new HashMap();

    public void execute(Tuple tuple, BasicOutputCollector collector) {

        // TODO Auto-generated method stub

        String word = tuple.getString(0);

        Integer count = counts.get(word);

        if (count == null)

            count = 0;

        count++;

        counts.put(word, count);

        for (Entry e : counts.entrySet()) {

            //sum += e.getValue();

            System.out.println(e.getKey()

                                + "----------->" +e.getValue());

        }

        collector.emit(new Values(word, count));     

    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        // TODO Auto-generated method stub

         declarer.declare(new Fields("word", "count"));

    }

}

package com.kafka.spout;

import java.util.HashMap;

import java.util.Map;

import com.google.common.collect.Maps;

//import org.apache.storm.guava.collect.Maps;

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.spout.SchemeAsMultiScheme;

import backtype.storm.topology.TopologyBuilder;

import backtype.storm.tuple.Fields;

import backtype.storm.utils.Utils;

import storm.kafka.BrokerHosts;

import storm.kafka.KafkaSpout;

import storm.kafka.SpoutConfig;

import storm.kafka.ZkHosts;

public class StormKafkaTopo {

    public static void main(String[] args) {

        BrokerHosts brokerHosts = new ZkHosts("zeb,yjd,ylh");

        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "yjd", "/storm", "kafkaspout");

        Config conf = new Config();  

        spoutConfig.scheme =  new SchemeAsMultiScheme(new MessageScheme());   

        SimpleHBaseMapper mapper = new SimpleHBaseMapper();

        mapper.withColumnFamily("result");

        mapper.withColumnFields(new Fields("count"));

        mapper.withRowKeyField("word");

        Map map = Maps.newTreeMap();

        map.put("hbase.rootdir", "hdfs://zeb:9000/hbase");

        map.put("hbase.zookeeper.quorum", "zeb:2181,yjd:2181,ylh:2181");

        // hbase-bolt

        HBaseBolt hBaseBolt = new HBaseBolt("wc", mapper).withConfigKey("hbase.conf");

        conf.setDebug(true);

        conf.put("hbase.conf", map);

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", new KafkaSpout(spoutConfig));

        builder.setBolt("split", new LevelSplit(), 1).shuffleGrouping("spout");

        builder.setBolt("count", new LevelCount(), 1).fieldsGrouping("split", new Fields("word"));

        builder.setBolt("hbase", hBaseBolt, 1).shuffleGrouping("count");

        if(args != null && args.length > 0) {

            //提交到集群运行

            try {

                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());

            } catch (AlreadyAliveException e) {

                e.printStackTrace();

            } catch (InvalidTopologyException e) {

                e.printStackTrace();

            }

        } else {

            //本地模式运行

            LocalCluster cluster = new LocalCluster();

            cluster.submitTopology("Topotest1121", conf, builder.createTopology());

            Utils.sleep(1000000);

            cluster.killTopology("Topotest1121");

            cluster.shutdown();

        }          

    }

}