基于Storm本地集群搭建实时统计CallLog实现可运行

package com.zj.storm; import org.apache.storm.spout.SpoutOutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.IRichSpout; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Random; /** * 产生数据源的 */ public class CallLogSpout implements IRichSpout { //Spout输出收集器 private SpoutOutputCollector collector; private boolean completed = false; private TopologyContext context; //上下文 private Random random = new Random(); private Integer idx = 0; @Override public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) { this.collector = spoutOutputCollector; this.context = topologyContext; } @Override public void close() { } @Override public void activate() { } @Override public void deactivate() { } /** * 下一个元组 * 一直运行 */ @Override public void nextTuple() { if (this.idx <= 1000) { List<String> mobileNumbers = new ArrayList<>(); mobileNumbers.add("1234123401"); mobileNumbers.add("1234123402"); mobileNumbers.add("1234123403"); mobileNumbers.add("1234123404"); Integer localIdx = 0; while(localIdx++ < 100 && this.idx++ < 1000) { String caller = mobileNumbers.get(random.nextInt(4)); String callee = mobileNumbers.get(random.nextInt(4)); while(caller == callee) { callee = mobileNumbers.get(random.nextInt(4)); } Integer duration = random.nextInt(60); //输出元组 this.collector.emit(new Values(caller, callee, duration)); } } } @Override public void ack(Object o) { } @Override public void fail(Object o) { } /** * 定义输出的字段名 * * @param outputFieldsDeclarer */ @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declare(new Fields("from", "to", "duration")); } @Override public Map<String, Object> getComponentConfiguration() { return null; } }