数据样本

  1. i am jdxia
  2. i am xjd
  3. i am jdxia
  4. i am jelly

jar包

  1. <dependency>
  2.     <groupId>org.apache.storm</groupId>
  3.     <artifactId>storm-core</artifactId>
  4.     <!--<scope>provided</scope>-->
  5.     <version>0.9.5</version>
  6. </dependency>

安装log4j

数据获取

BaseRichSpout类是ISpout接口和IComponent接口的一个简便的实现

open方法中接收三个参数

  • conf包含了storm配置信息的map.
  • TopologyContext对象提供了topology中组件的信息
  • SpoutOutputCollector对象提供了发射tuple的方法
  1. package com.learnstorm;
  3. import backtype.storm.spout.SpoutOutputCollector;
  4. import backtype.storm.task.TopologyContext;
  5. import backtype.storm.topology.OutputFieldsDeclarer;
  6. import backtype.storm.topology.base.BaseRichSpout;
  7. import backtype.storm.tuple.Fields;
  8. import org.apache.commons.lang.StringUtils;
  10. import java.io.*;
  11. import java.util.ArrayList;
  12. import java.util.List;
  13. import java.util.Map;
  15. //数据获取
  16. public class MyLocalFileSpout extends BaseRichSpout {
  18.     //控制数据输出
  19.     private SpoutOutputCollector collector;
  20.     //读取数据的
  21.     private BufferedReader bufferedReader;
  23.     //初始化方法
  24.     @Override
  25.     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
  26.         this.collector = collector;
  27.         try {
  28.             //定义这个去读取数据
  29.             this.bufferedReader = new BufferedReader(new FileReader(new File("/Users/jdxia/Desktop/MyFile/i.txt")));
  30.         } catch (FileNotFoundException e) {
  31.             e.printStackTrace();
  32.         }
  33.     }
  35.     //storm流式计算的特征就是数据一条一条的处理
  36. //    while(true) {
  37. //        this.nextTuple();
  38. //    }
  39.     //这个方法会被循环调用
  40.     @Override
  41.     public void nextTuple() {
  42.         //每被调用一次就会发送一条数据出去
  43.         try {
  44.             //读取一行
  45.             String line = bufferedReader.readLine();
  46.             //如果不是空的话
  47.             if (StringUtils.isNotBlank(line)) {
  48.                 List<Object> arrayList = new ArrayList<Object>();
  49.                 //把数据放到ArrayList中
  50.                 arrayList.add(line);
  51.                 //把数据发出去
  52.                 collector.emit(arrayList);
  53.             }
  54.         } catch (IOException e) {
  55.             e.printStackTrace();
  56.         }
  57.     }
  59.     //定义下我的输出
  60.     @Override
  61.     public void declareOutputFields(OutputFieldsDeclarer declarer) {
  62.         declarer.declare(new Fields("juzi"));
  63.     }
  64. }

数据截取

BaseBasicBolt是IComponent和IBolt接口的一个简便实现
BaseBasicBolt中还有个prepare()方法,是bolt初始化的时候调用的

  1. package com.learnstorm;
  3. import backtype.storm.topology.BasicOutputCollector;
  4. import backtype.storm.topology.OutputFieldsDeclarer;
  5. import backtype.storm.topology.base.BaseBasicBolt;
  6. import backtype.storm.tuple.Fields;
  7. import backtype.storm.tuple.Tuple;
  8. import backtype.storm.tuple.Values;
  10. //相当于map-->world,1
  11. //业务逻辑
  12. //对句子进行切割
  13. public class MySplitBolt extends BaseBasicBolt {
  15.     //处理函数
  16.     @Override
  17.     public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
  18.         //1.数据如何获取,用tuple获取,tuple是List数据结构,消息传输的基本单元
  19.         //强转为string,juzi是上一步定义的
  20.         String juzi = (String) tuple.getValueByField("juzi");
  22.         //2.进行切割
  23.         String[] strings = juzi.split(" ");
  25.         //3.发送数据
  26.         for (String word : strings) {
  27.             //我们之前用ArrayList存储,这边怎么变为Values
  28.             //可以看下Values的源码,他是继承了ArrayList,他存的时候用了一个循环
  29.             //values对象帮我们生成个list
  30.             basicOutputCollector.emit(new Values(word, 1));
  31.         }
  32.     }
  34.     //定义下我的输出
  35.     //单词world和他的次数
  36.     @Override
  37.     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  38.         outputFieldsDeclarer.declare(new Fields("word", "num"));
  39.     }
  40. }

单词统计

这里面用到了HashMap.这是可序列化的
如果spout或者bolt在序列化之前,(比如在构造函数中生成)实例化了任何无法序列化的实例变量,在进行序列化的时候会抛出NotSerialiableException

当topology发布时,所有的bolt和spout组件首先进行序列化,然后发布到网络中.

通常情况下最好在构造函数中对基本数据类型和可序列化的对象进行赋值和实例化,在prepare()方法中对不可序列化的对象进行实例化

bolt中可以加入cleanup()方法,这个方法在IBolt中定义.storm在终止一个Bolt之前会调用这个方法.通常情况下cleanup会用来释放bolt占用的资源
当在集群中运行的时候,cleanup是不可靠的,不能保证会执行

  1. package com.learnstorm;
  3. import backtype.storm.topology.BasicOutputCollector;
  4. import backtype.storm.topology.OutputFieldsDeclarer;
  5. import backtype.storm.topology.base.BaseBasicBolt;
  6. import backtype.storm.tuple.Tuple;
  8. import java.util.HashMap;
  9. import java.util.Map;
  11. //打印
  12. public class MyWordCountAndPrintBolt extends BaseBasicBolt {
  14.     private Map<String, Integer> wordCountMap = new HashMap<String, Integer>();
  16.     //处理函数
  17.     @Override
  18.     public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
  19.         //根据之前定义的word和num
  20.         //强转为string
  21.         String word = (String) tuple.getValueByField("word");
  22.         Integer num = (Integer) tuple.getValueByField("num");
  24.         //1.查看单词对应的value是否存在
  25.         Integer integer = wordCountMap.get(word);
  26.         if (integer == null || integer.intValue() == 0) {
  27.             //如果不存在就直接放入新的
  28.             wordCountMap.put(word, num);
  29.         } else {
  30.             //如果之前已经有了,就把对应统计加上
  31.             wordCountMap.put(word, integer.intValue() + num);
  32.         }
  34.         System.out.println(wordCountMap);
  35.     }
  37.     //不需要定义输出字段了
  38.     @Override
  39.     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
  41.     }
  43.     //终止bolt会调用这个方法
  44.     @Override
  45.     public void cleanup() {
  47.     }
  48. }

任务描述

这边写的是本地提交到集群

  1. package com.learnstorm;
  4. import backtype.storm.Config;
  5. import backtype.storm.LocalCluster;
  6. import backtype.storm.StormSubmitter;
  7. import backtype.storm.generated.AlreadyAliveException;
  8. import backtype.storm.generated.InvalidTopologyException;
  9. import backtype.storm.generated.StormTopology;
  10. import backtype.storm.topology.TopologyBuilder;
  12. public class StormTopologyDriver {
  14.     public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
  16.         //1. 描述任务
  17.         TopologyBuilder topologyBuilder = new TopologyBuilder();
  18.         //任务的名字自己定义
  19.         topologyBuilder.setSpout("mySpout", new MyLocalFileSpout());
  20.         //shuffleGrouping和前一个任务关联.shuffleGrouping可以连接多个任务
  21.         topologyBuilder.setBolt("bolt1", new MySplitBolt()).shuffleGrouping("mySpout");
  22.         topologyBuilder.setBolt("bolt2", new MyWordCountAndPrintBolt()).shuffleGrouping("bolt1");
  24.         //2. 任务提交
  25.         //提交给谁?提交什么内容?
  26.         Config config = new Config();   //Config类实际上是继承HashMap
  27.         //设置在几个work上运行,也就是在几个jvm中运行,如果不指定,默认是在一个work中
  28. //        config.setNumWorkers(2);
  29.         StormTopology stormTopology = topologyBuilder.createTopology();
  31.         //本地模式
  32.         LocalCluster localCluster = new LocalCluster();
  33.         localCluster.submitTopology("wordCount", config, stormTopology);
  35.         //这种是集群模式
  36. //        StormSubmitter.submitTopology("worldCount1", config, stormTopology);
  38.     }
  39. }

提交到集群

如果是提交到集群上面,那么storm的storm-core的作用域就要改下

  1. <scope>provided</scope>

表示集群上提供了这个jar包,
然后maven对项目打包,上传到服务器上,执行

  1. storm jar 上传的jar包 主类名称

然后我们在ui界面上看

image.png

点进去看

image.png

统计信息打印在bolt2上

查看下bolt2在那台机器上

image.png

hadoop03的6700端口上

到这台机器上的storm的logs目录
目录下有个worker-6700.log查看下这个日志