两个例子(来自Storm实战 构建大数据实时计算)

转自http://blog.csdn.net/wust__wangfan/article/details/50517554

例子一：模拟网站计算用户PV(页面浏览量) 拓扑图如下：

1、编写Topology [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class TopoMain {

public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException { TopologyBuilder builder=new TopologyBuilder(); builder.setSpout("log-reader", new LogReader(),1); builder.setBolt("log-stat", new LogStat(), 2).fieldsGrouping("log-reader",new Fields("user")); builder.setBolt("log-writer", new LogWriter(),1).shuffleGrouping("log-stat"); Config config=new Config(); config.setNumWorkers(5); StormSubmitter.submitTopology("log-topology", config, builder.createTopology()); }

}

2、编写spout [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogReader extends BaseRichSpout{

private SpoutOutputCollector _collector; private Random _rand=new Random(); private int _count=100; private String[] _users={"userA","userB","userC","userD","userE"}; private String[] _urls={"url1","url2","url3","url4","url5"}; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { _collector=collector; } @Override public void nextTuple() { try { Thread.sleep(1000); while(_count-->=0) { _collector.emit(new Values(System.currentTimeMillis(),_users[_rand.nextInt(5)],_urls[_rand.nextInt(5)])); } } catch (InterruptedException e) { e.printStackTrace(); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("time","user","url")); }

}

3、编写bolt [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogStat extends BaseRichBolt{

private OutputCollector _collector;//注意和SpoutOutputCollector区分 private Map<String, Integer>_pvMap=new HashMap<String,Integer>(); @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { _collector=collector; } @Override public void execute(Tuple input) { String user=input.getStringByField("user"); if(_pvMap.containsKey(user)) { _pvMap.put(user, _pvMap.get(user)+1); } else { _pvMap.put(user, 1); } //把每个用户的最新PV输出到下一节点 _collector.emit(new Values(user,_pvMap.get(user))); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("user","pv")); }

}

[java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogWriter extends BaseRichBolt{

private FileWriter writer=null; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { // TODO 自动生成的方法存根 try { writer=new FileWriter("/"+this); } catch (IOException e) { // TODO 自动生成的 catch 块 e.printStackTrace(); } } @Override public void execute(Tuple input) { try { writer.write(input.getStringByField("user")+" : "+input.getIntegerByField("pv")); writer.write("\n"); writer.flush(); } catch (IOException e) { e.printStackTrace(); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { }

} 对结果分析： 结果会输出最新的PV，如userA: 1, userB:1, userA:2, userA:3……………..

例子二、改进上面的例子,让最终结果每个用户只输出一次页面访问量

1、topology

[java] view plain copy 在CODE上查看代码片派生到我的代码片 public class TopoMain {

public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException { TopologyBuilder builder=new TopologyBuilder(); builder.setSpout("log-reader", new LogReader(),1); //流log负责输出日志，采用字段分组；流stop告诉下游输出完毕，全局广播这个tuple builder.setBolt("log-stat", new LogStat(), 2).fieldsGrouping("log-reader", "log", new Fields("user")).allGrouping("log-reader", "stop"); builder.setBolt("log-writer", new LogWriter(),1).shuffleGrouping("log-stat"); Config config=new Config(); config.setNumWorkers(5); StormSubmitter.submitTopology("log-topology", config, builder.createTopology()); }

}

2、spout [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogReader extends BaseRichSpout{

private SpoutOutputCollector _collector; private Random _rand=new Random(); private int _count=100; private String[] _users={"userA","userB","userC","userD","userE"}; private String[] _urls={"url1","url2","url3","url4","url5"}; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { _collector=collector; } @Override public void nextTuple() { try { Thread.sleep(1000); while(_count-->=0) { if(_count==0) { //发送""发给下流bolt _collector.emit("stop", new Values("")); } _collector.emit("log",new Values(System.currentTimeMillis(),_users[_rand.nextInt(5)],_urls[_rand.nextInt(5)])); } } catch (InterruptedException e) { e.printStackTrace(); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream("log",new Fields("time","user","url")); declarer.declareStream("stop", new Fields("flag")); }

}

3、bolt [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogStat extends BaseRichBolt{

private OutputCollector _collector;//注意和SpoutOutputCollector区分 private Map<String, Integer>_pvMap=new HashMap<String,Integer>(); @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { _collector=collector; } @Override public void execute(Tuple input) { String streamId=input.getSourceStreamId(); if(streamId.equals("log")) { String user=input.getStringByField("user"); if(_pvMap.containsKey(user)) { _pvMap.put(user, _pvMap.get(user)+1); } else { _pvMap.put(user, 1); } } if(streamId.equals("stop")) { //从map取数据发送 Iterator<Entry<String,Integer>> it=_pvMap.entrySet().iterator(); while(it.hasNext()) { Entry<String,Integer> entry=it.next(); _collector.emit(new Values(entry.getKey(),entry.getValue())); } } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("user","pv")); }

} [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogWriter extends BaseRichBolt{

private FileWriter writer=null; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { // TODO 自动生成的方法存根 try { writer=new FileWriter("/"+this); } catch (IOException e) { // TODO 自动生成的 catch 块 e.printStackTrace(); } } @Override public void execute(Tuple input) { try { writer.write(input.getStringByField("user")+" : "+input.getIntegerByField("pv")); writer.write("\n"); writer.flush(); } catch (IOException e) { e.printStackTrace(); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { }

}

例三、模拟一个简化的电子商务网站来实时计算来源成交效果

1、topology [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class TopoMain {

public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException { TopologyBuilder builder=new TopologyBuilder(); builder.setSpout("log-vspout", new VSpout(),1); builder.setSpout("log-bspout", new BSpout(),1); //定义两个流名称，分别为visit和business builder.setBolt("log-merge", new LogMergeBolt(),2).fieldsGrouping("log-vspout", "visit",new Fields("user")).fieldsGrouping("log-bspout","business",new Fields("user")); builder.setBolt("log-stat", new LogStatBolt(),2).fieldsGrouping("log-merge",new Fields("srcid")); builder.setBolt("log-writer", new LogWriter()).globalGrouping("log-stat"); Config conf=new Config(); //实时计算不需要可靠消息，故关闭acker节省通信资源 conf.setNumAckers(0); //设置独立java进程数，一般设置为同spout和bolt的总tasks数量相等或更多，使得每个task //都运行在独立的Java进程中，以避免多个task集中在一个jvm里运行产 //生GC(垃圾回收机制)瓶颈 conf.setNumWorkers(8); StormSubmitter.submitTopology("ElectronicCom_Top", conf, builder.createTopology()); }

}

2、spout [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class BSpout extends BaseRichSpout{

private SpoutOutputCollector _collector; private String[] _users={"userA","userB","userC","userD","userE"}; private String[] _pays={"100","200","300","400","200"}; private int count=5; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { _collector=collector; } @Override public void nextTuple() { for(int i=0;i<count;i++) { try { Thread.sleep(1500);//停顿时间长一些，使得流量日志先于成交日志到达下游的LogMergeBolt组件 _collector.emit("business", new Values(System.currentTimeMillis(),_users[i],_pays[i])); } catch (InterruptedException e) { e.printStackTrace(); } } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream("business", new Fields("time","user","pay")); }

}

[java] view plain copy 在CODE上查看代码片派生到我的代码片 public class VSpout extends BaseRichSpout{

private SpoutOutputCollector _collector; private String[] _users={"userA","userB","userC","userD","userE"}; private String[] _srcid={"s1","s2","s3","s1","s1"}; private int count=5; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { _collector=collector; } @Override public void nextTuple() { for(int i=0;i<count;i++) { try { Thread.sleep(1000); _collector.emit("visit", new Values(System.currentTimeMillis(),_users[i],_srcid[i])); } catch (InterruptedException e) { e.printStackTrace(); } } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream("visit", new Fields("time","user","srcid")); }

}

3、bolt [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogMergeBolt extends BaseRichBolt{

private OutputCollector _collector; //暂时存储用户的访问日志记录 private HashMap<String, String> srcMap; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { _collector=collector; if(srcMap==null) { srcMap=new HashMap<String, String>(); } } @Override public void execute(Tuple input) { String streamID=input.getSourceStreamId(); if(streamID.equals("visit")) { String user=input.getStringByField("user"); String srcid=input.getStringByField("srcid"); srcMap.put(user, srcid); } if(streamID.equals("business")) { String user=input.getStringByField("user"); String pay=input.getStringByField("pay"); String srcid=srcMap.get(user); if(srcid!=null) { _collector.emit(new Values(user,pay,srcid)); srcMap.remove(user); } else { //一般只有成交日志先于流量日志才会发生 } } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("user","srcid","pay")); }

}

[java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogStatBolt extends BaseRichBolt{

private OutputCollector _collector; private HashMap<String, Long> srcpay; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { _collector=collector; if(srcpay==null) { srcpay=new HashMap<String, Long>(); } } @Override public void execute(Tuple input) { String pay=input.getStringByField("pay"); String srcid=input.getStringByField("srcid"); if(srcpay.containsKey(srcid)) { srcpay.put(srcid, srcpay.get(srcid)+Long.parseLong(pay.trim())); } else { srcpay.put(srcid, Long.parseLong(pay.trim())); } _collector.emit(new Values(srcid,srcpay.get(srcid))); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("srcid","pay")); }

} [java] view plain copy 在CODE上查看代码片派生到我的代码片 public class LogWriter extends BaseRichBolt{ private HashMap