package com.bjsxt.storm.wc;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;

public class WordCountTopology {

    public static void main(String[] args) {
        // 拓扑封装了计算逻辑
        TopologyBuilder builder = new TopologyBuilder();
        // 设置lineSpout：该spout负责向拓扑发送句子
        builder.setSpout("lineSpout", new LineSpout());
        // 设置切分闪电，该闪电处理从水龙头lineSpout通过随机分组发送过来的元组
        builder.setBolt("splitBolt", new SplitBolt())
            .shuffleGrouping("lineSpout");
        // 定义一个计数闪电，该闪电从splitBolt闪电通过按字段分组的方式分发过来的元组
        // 按照元组中word的值进行分组。要保证相同的单词一定发送给同一个闪电。
        builder.setBolt("countBolt", new CountBolt())
                .fieldsGrouping("splitBolt", new Fields("word"));

        // 通过建造者创建一个拓扑的实例
        StormTopology wordCountTopology = builder.createTopology();

        // 本地模拟集群
        LocalCluster cluster = new LocalCluster();

        Config config = new Config();

        // 将拓扑提交到本地模拟集群
        cluster.submitTopology("wordCountTopology", config, wordCountTopology);

        // 睡眠10s，也就是让本地模拟集群运行10s
        Utils.sleep(10000);

        // 关闭本地模拟集群
        cluster.shutdown();

    }

}

package com.bjsxt.storm.wc;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

import java.util.Map;

public class LineSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;

    private String[] lines = {
            "The logic for a realtime application is packaged into a Storm topology",
            "A stream is an unbounded sequence of tuples that is processed and created in parallel in a distributed fashion",
            "A spout is a source of streams in a topology",
            "Bolts can do anything from filtering, functions, aggregations, joins, talking to databases, and more.",
            "A stream grouping defines how that stream should be partitioned among the bolt's tasks.",
            "Storm guarantees that every spout tuple will be fully processed by the topology",
            "Each spout or bolt executes as many tasks across the cluster",
            "Each worker process is a physical JVM and executes a subset of all the tasks for the topology"
    };

    private int index = 0;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        // 在该组件在集群中初始化的时候调用一次
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // 由storm的线程不停地调用，以便从数据源获取元组
        // 该方法不需要自己写循环和遍历
        // 该方法不能阻塞
        // 负责从数据源获取元组，向DAG发送元组
        // 轮询取出句子
        String lingStr = lines[index % lines.length];
        // 将句子封装为元组发射
        collector.emit(new Values(lingStr));
        index++;

        Utils.sleep(10);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 用于声明元组的结构以及流
//        declarer.declareStream("s1", new Fields("key1", "key2", "key3"));
//        declarer.declareStream("s2", new Fields("key21", "key22"));
        // 发送元组的时候就有一个字段，是line，它的值是句子
        // 可以将元组想象为map集合，只不过其key是固定的几个
        declarer.declare(new Fields("line"));
    }
}

package com.bjsxt.storm.wc;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import java.util.Map;

public class SplitBolt extends BaseRichBolt {

    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String lineStr = input.getStringByField("line");
        String[] wordStrs = lineStr.split(" ");

        for (String wordStr : wordStrs) {
            // <hello, 1>
            this.collector.emit(new Values(wordStr, 1));
        }

    }

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

package com.bjsxt.storm.wc;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

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

public class CountBolt extends BaseRichBolt {

    private Map<String, Integer> counts;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        counts = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
//        new Fields("word", "count")
        String wordStr = input.getStringByField("word");
        Integer count = input.getIntegerByField("count");

        Integer sum = counts.get(wordStr);

        if (sum == null) {
            counts.put(wordStr, count);
        } else {
            counts.put(wordStr, sum + count);
        }

        counts.forEach((k, v) -> {
            System.out.println(k + "_________" + v);
        });
        System.out.println("========================================");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}