02 Stormのローカルモードを使用して語周波数統計を完了

前にstormをインストールしました.stormには2つのモードがあります.1つはローカルモードで、主に学習とテストに使用され、もう1つはクラスタモードで、実際の生産ではこのモードが使用されています.このセクションでは、ローカルモードのstormを使用して語周波数統計を行う方法について説明します.
1システム、ソフトウェアおよび前提条件

CentOS 7 64ワークステーションの作者のマシンipは192.168.100.200で、読者は自分の実際の状況によって

を設定してください.

idea 2018.1

2操作

1 ideaでmavenプロジェクト

を作成

2 pom.xmlを変更し、以下の依存

を加える.

    
        
            
            org.apache.spark
            spark-core_2.11
            2.2.0
        
        
        
            org.scala-lang
            scala-library
            2.11.8
        
        
        
            org.apache.hadoop
            hadoop-client
            2.6.0-cdh5.7.0
        
        
        
            org.apache.hbase
            hbase-client
            2.0.0-cdh6.0.1
        
        
        
            org.apache.storm
            storm-core
            
                
                    org.slf4j
                    log4j-over-slf4j
                
            
            1.2.1
        
    

jarパッケージのダウンロードが完了するのを待っています.

3 src/main/javaにRandomSentenceSpout.javaをデータソースとして追加

import java.util.Map;
import java.util.Random;

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;

public class RandomSentenceSpout extends BaseRichSpout {
    SpoutOutputCollector _collector;
    Random _rand;

    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        _collector = collector;
        _rand = new Random();
    }

    public void nextTuple() {
        String[] sentences = new String[] { "the cow jumped over the moon", "an apple a day keeps the doctor away" };
        String sentence = sentences[_rand.nextInt(sentences.length)];
        _collector.emit(new Values(sentence));
    }

    public void ack(Object id) {
    }

    public void fail(Object id) {
    }

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

4 src/main/javaにSplitSentenceBolt.javaを加えてデータ分割

を行う

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

public class SplitSentenceBolt extends BaseBasicBolt {

    private static final long serialVersionUID = -1L;

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

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

5 src/main/javaにWordCountBolt.javaを単語統計として追加

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

public class WordCountBolt extends BaseBasicBolt {
    private static final long serialVersionUID = -1L;
    private Map counts = new HashMap();


    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String word = tuple.getString(0);
        Integer count = counts.get(word);
        if (count == null) {
            count = 0;
        }
        count++;
        counts.put(word, count);
        System.out.println(Thread.currentThread().getId() + "=========== word : " + word + " count: " + count);
    }

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

6 src/main/javaにWordCountSub.javaを追加してプロセス制約とトポロジコミット

を行います.

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class WordCountSub{
    public static void main(String[] args) throws Exception {

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", new RandomSentenceSpout(), 5);

        builder.setBolt("split", new SplitSentenceBolt(), 8).shuffleGrouping("spout");
        builder.setBolt("count", new WordCountBolt(), 12).fieldsGrouping("split", new Fields("word"));

        Config conf = new Config();
        conf.setNumWorkers(3);
        conf.setNumAckers(1);
        LocalCluster localCluster= new LocalCluster();
        localCluster.submitTopology("test",conf,builder.createTopology());
    }

}

7マウスの右ボタンでWordCountSub.javaを実行し、「長いかもしれませんが、このプロセスはメモリが消費されます」としばらく待っていると、コンソールに出力が表示されます.以上,stormのローカルモードを用いて語周波数統計を行った.