大数据课程——Storm综合应用

大数据课程——Storm综合应用实验内容以及要求

假设在某一搜索应用中，需要实时统计搜索次数最多的热门关键词，并按照实时统计搜索次数输出最热门的20大热门关键词及被搜索次数。用户搜索的日志通过Flume采集，并写入Kafka，Storm从Kafka中实时读取数据，并完成实时统计工作，并进行输出。

提示：
（1）搜索日志可以采用实验5的数据（搜狗搜索数据），一行代表一次用户搜索；
（2）Flume Agent的Source可以配置为syslogudp类型（端口5640），监控搜索日志；
（3）输出形式自定。

问题总结 Agent配置时的Source类型

Client包是模拟数据包的产生的，将数据发往5640端口。因此Flume的Agent配置要写明Source是syslogudp类型，并且监控5640端口。

Storm的UI界面端口问题

Storm的UI界面端口是可以改动的，默认为8080。但在实验室里做实验的话，就发现老师已经吧端口改了，改成了8099，因此要访问localhost:8099才可以进入Storm的UI界面，否则会出现404错误，这个要注意。

Key、Value问题

写代码的时候，一开始照着书上的Kafka整合Storm的代码来写，但其实获取数据的时候，其实Key是没东西的，Value才能获得具体数据。Key得到的全是null。如下图所示。

吐槽一下

顺便吐槽一下，实验数据输出的排行榜略少儿不宜。但没办法，只能说数据太真实了哈哈哈。
另外具体代码放在本文最后面。

实验步骤 Flume Agent 配置

本次将Centos01作为Flume的Agent，负责监控5640端口，拦截收到的数据包，并将其中的数据存储到Kafka集群的topictest话题中。
所以将Source的类型设置为syslogudp，监控5640端口，接受其中产生的数据包。
Sink设置为Kafka集群，将得到的数据包传入kafka集群中。

Storm代码编写思路

如上图所示，我的项目中主体由KafkaSpout、SplitDataBolt、WordCountBolt、ReportBolt四项组成，数据在其中依次流动，StormTopology对他们进行设置、调用。（代码在文末）。
KafkaSpout作为数据源，从kafka集群中提取需要的数据，并且发送给SplitDataBolt进行处理。
SplitOutBolt作为一个处理Bolt，将接收到的数据进行分割处理，从中提取出关键字字段，并且设置值为1（因为出现一次），发送给WordCountBolt进行统计。
WordCountBolt作为第二个处理Bolt，对关键词的出现次数进行统计和存储，并将统计结果发送给ReportBolt。
ReportBolt作为第三个Bolt，将接收到的“关键词-出现次数”数据进行存储，并且按照“出现次数”进行排序，将排序结果进行输出，得到最终的结果排行榜。

代码展示

首先依次开启Zookeeper、HDFS、HA集群、Kafka、Flume、Storm。并且可以进入Storm的UI界面来测试Storm是否开启成功，其他的集群开启这里不予以赘述。

系统开启完毕后，就可以运行Client.jar包，该JAR包的作用是负责从数据集中提取数据，整合成数据包，并且随机间隔一定时间向5640端口发送该数据包。这样就可以模拟真实的数据产生，并且被Flume抓住该数据包，发送到Kafka集群中。

如果一切顺利的话，可以开启一个消费者，查看topictest话题，能看到数据正常被发送到Kafka集群中。如下图所示。

项目代码写好后，打包成Storm.jar包，在Centos01中运行，成功运行后如下图所示。

每间隔一段时间，都会打印出排行榜信息，如下图所示。可以看到随着时间的增加，排行榜是在根据数据不断变化的，不断输出排名前20的关键词排行信息，可以看到关键词以及其出现次数统计。

代码 KafkaSpout

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
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 java.time.Duration;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;


public class KafkaSpout extends BaseRichSpout {
    private static final long serialVersionUID = 7582771881226024741L;
    private KafkaConsumer consumer;
    SpoutOutputCollector collector;


    
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
        //Kafka属性信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "centos01:9092");
        props.put("group.id", "test");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //实例化消费者，设置消费主题
        consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList("topictest"));
    }
    
    public void nextTuple() {
        ConsumerRecords records = consumer.poll(Duration.ofSeconds(10));
        for (ConsumerRecord record : records) {
            String key = record.key();
            String value = record.value();
            //其实KEY值是没有的，Value值才是有的
            //所以这里吧value的值赋给了Key
            key = value;
            System.out.println("key " + key);
//            System.out.println("value " + value);

            collector.emit(new Values(key));
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("key"));
    }
}

SplitDataBolt

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 org.apache.storm.tuple.Values;

import java.util.Map;


public class SplitDataBolt extends BaseRichBolt {
    private static final long serialVersionUID = 1L;
    private OutputCollector outputcollector;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputcollector = outputCollector;
    }

    
    @Override
    public void execute(Tuple tuple) {
        String key =tuple.getStringByField("key");
        //分割数据
        String[] words =key.split("t");
        //提取关键词字段
        String keyword = words[2].trim();
        Integer cnt = 1;
        //发送给下一bolt
        this.outputcollector.emit(new Values(keyword, cnt));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("keyword", "cnt"));
    }
}

WordCountBolt

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 org.apache.storm.tuple.Values;

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


public class WordCountBolt extends BaseRichBolt {
    private static final long serialVersionUID = 2374950653902413273L;
    private OutputCollector outputcollector;


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputcollector = outputCollector;
        this.keywordMap = new HashMap();
    }



    
    //定义存放单词与词频的Map
    private HashMap keywordMap = null;
    @Override
    public void execute(Tuple tuple) {
        String keyword = tuple.getStringByField("keyword");
        int cnt = tuple.getIntegerByField("cnt");
        Integer sum = keywordMap.get(keyword);
        if (sum == null) {
            sum = 0;
        }
        sum++;
        keywordMap.put(keyword, sum);
        this.outputcollector.emit(new Values(keyword, sum));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("keyword", "sum"));
    }
}

ReportBolt

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.Tuple;

import java.util.*;
import java.util.Map.Entry;


public class ReportBolt extends BaseRichBolt {
    private static final long serialVersionUID = -1512537746316594950L;


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.keywordMap = new HashMap();
    }

    
    private HashMap keywordMap = null;
    @Override
    public void execute(Tuple tuple) {
        String keyword = tuple.getStringByField("keyword");
        int count = tuple.getIntegerByField("sum");
        keywordMap.put(keyword, count);
        List> list = new ArrayList>(keywordMap.entrySet());
        Collections.sort(list, new Comparator>() {
            //升序排列
            @Override
            public int compare(Entry o1, Entry o2) {
                return (o2.getValue() - o1.getValue());
            }
        });
        //输出排名前20的数据
        int n = list.size() <= 20 ? list.size() : 20;
        String result = "";
        for (int i = 0; i < n; i++) {
            Entry entry = list.get(i);
            String sortWord = entry.getKey();
            Integer sortCount = entry.getValue();
            result += sortWord + " ------- " + sortCount +"n";
        }
        System.out.println("--------(关键词)--------搜索关键词排行榜--------(搜索次数)--------");
        System.out.println(result);

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}

StormTopology

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 StormTopology {
    public static void main(String[] args) throws Exception {
        KafkaSpout kafkaSpout = new KafkaSpout();
        SplitDataBolt splitDataBolt = new SplitDataBolt();
        WordCountBolt wordCountBolt = new WordCountBolt();
        ReportBolt reportBolt = new ReportBolt();

        //创建一个拓扑
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //TODO 这些命名还有fields的参数可能会出问题，可以看看
        //设置Spout，名称为"kafka-spout"，并行度为2(也就是线程数)，任务数为4(也就是实例数)。默认是1个线程，1个任务。  如果不指定Task数量，则一个线程执行一个Task，Task数量与线程数量一样。
        topologyBuilder.setSpout("kafka-spout", kafkaSpout,2).setNumTasks(4);
        //设置bolt，名称为"split-bolt",数据来源是名称为"sentence-spout"的spout，
        //ShuffleGrouping：随机选择一个Task来发送,对Task的分配比较均匀。
        topologyBuilder.setBolt("split-bolt", splitDataBolt,2).setNumTasks(4).shuffleGrouping("kafka-spout");
        //FiledGrouping：根据Tuple中Fields来做一致性hash，相同hash值的Tuple被发送到相同的Task。
        topologyBuilder.setBolt("count-bolt", wordCountBolt,2).setNumTasks(4).fieldsGrouping("split-bolt", new Fields("keyword"));
        //GlobalGrouping：所有的Tuple会被发送到某个Bolt中的id最小的那个Task,此时不管有多少个Task，只发往一个Task
        topologyBuilder.setBolt("report-bolt", reportBolt,2).setNumTasks(4).globalGrouping("count-bolt");

        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        //本地模式 ，第一个参数为定义拓扑名称
//      cluster.submitTopology("word-count-topology", config, topologyBuilder.createTopology());
       

        //集群模式，需要打包jar上传到集群，然后使用命令 :storm jar storm_demo-0.0.1-SNAPSHOT.jar com.zwy.storm.demo.wordcount.WordCountTopology
        config.setNumWorkers(2); //设置Worker进程数量
        config.setNumAckers(0);//设置acker并发数，0代表取消acker任务。Acker任务默认是每个worker进程启动一个executor线程来执行,该实例启动了2个worker，则默认会启动2个executor线程,2个acker任务
        StormSubmitter.submitTopology("keyword-rank-topology",config,topologyBuilder.createTopology());
    }
}

大数据课程——Storm综合应用

Java相关栏目本月热门文章