storm详解

1.  构建拓扑代码

package demo;import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;public class AreaAmtTopo {public static void main(String[] args) {TopologyBuilder builder &＃61; new TopologyBuilder();
builder.setSpout("spout", new OrderBaseSpout(KafkaProperties.Order_topic),5);
builder.setBolt("filter",new AreaFilterBolt(),5).shuffleGrouping("spout");
builder.setBolt("areabolt",new AreaAmtBolt(),2).fieldsGrouping("filter",new Fields("area_id"));
builder.setBolt("rsltbolt",new AreaRsltBolt(),1).shuffleGrouping("areabolt");}}

2.一级过滤bolt

package demo;import java.util.Map;import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
//一级的过滤bolt
public class AreaFilterBolt implements IBasicBolt {&＃64;Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stubdeclarer.declare(new Fields("area_id","order_amt","create_time"));//tuple里面每个value的对应name}&＃64;Overridepublic Map getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}&＃64;Overridepublic void cleanup() {// TODO Auto-generated method stub}&＃64;Overridepublic void execute(Tuple input, BasicOutputCollector collector) {//order_id,order_amt,create_time,area_idString order&＃61;input.getString(0);//取出集合values中的第一个valueif(order!&＃61;null){String orderArr[]&＃61;order.split("\\t");collector.emit(new Values(orderArr[3],orderArr[1],DateFmt.getCountDate(orderArr[2], DateFmt.date_short)));//area_id,order_amt,create_time}}&＃64;Overridepublic void prepare(Map arg0, TopologyContext arg1) {// TODO Auto-generated method stub}}

3.局部汇总bolt(按日期和区域和汇总&＃xff09;

package demo;import java.util.HashMap;
import java.util.Map;import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;//局部汇总
public class AreaAmtBolt implements IBasicBolt {Map countsMap&＃61;null;&＃64;Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("date_area","amt"));}&＃64;Overridepublic Map getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}&＃64;Overridepublic void prepare(Map paramMap, TopologyContext paramTopologyContext) {// TODO Auto-generated method stubcountsMap &＃61;new HashMap();}&＃64;Overridepublic void execute(Tuple input,BasicOutputCollector collector) {if(input!&＃61;null)//如果spout端没数据就会发空值,所以要做判断再往下发{String area_id&＃61;input.getString(0);Double order_amt&＃61;input.getDouble(1);String  order_date&＃61;input.getStringByField("order_date");Double count&＃61;countsMap.get(area_id&＃43;"_"&＃43;order_date);if (count&＃61;&＃61;null){count &＃61; 0.0;    }count&＃43;&＃61;order_amt;countsMap.put(area_id&＃43;"_"&＃43;order_date,count);System.err.println("areaAmtBolt"&＃43;order_date&＃43;"_"&＃43;area_id&＃43;"&＃61;"&＃43;count);collector.emit(new Values(area_id&＃43;"_"&＃43;order_date,count));}}&＃64;Overridepublic void cleanup() {countsMap.clear();}}

4. 最终结果写入Hbase

package demo;import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;//结果定时写入hbase的bolt
public class AreaRsltBolt implements IBasicBolt {Map countsMap&＃61;null;long beginTime&＃61;System.currentTimeMillis();long endTime&＃61;0L;HBaseDao dao&＃61;null;&＃64;Overridepublic void declareOutputFields(OutputFieldsDeclarer paramOutputFieldsDeclarer) {// TODO Auto-generated method stub}&＃64;Overridepublic Map getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}&＃64;Overridepublic void prepare(Map paramMap, TopologyContext paramTopologyContext) {countsMap &＃61;new HashMap();dao&＃61;new HBaseDAOImp();}&＃64;Overridepublic void execute(Tuple input,BasicOutputCollector paramBasicOutputCollector) {String date_areaid&＃61;input.getString(0);double  order_amt&＃61;input.getDouble(1); countsMap.put(date_areaid,order_amt);endTime&＃61;System.currentTimeMillis();if (endTime-beginTime>&＃61;5*1000){for(String key:countsMap.keySet()){//put into hbase//2014-05-05_1,amtdao.insert("area_order","cf","order_amt",countsMap.get(key));System.err.println("rsltBolt put hbase: key&＃61;"&＃43;key&＃43;"; order_amt&＃61;"&＃43;countsMap.get(key));}beginTime&＃61;System.currentTimeMillis();}}&＃64;Overridepublic void cleanup() {// TODO Auto-generated method stub}}

5. DateFmt代码

package demo;import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;public class DateFmt {public static final String date_long&＃61;"yyyy-MM-dd HH:mm:ss";public static final String date_short&＃61;"yyyy-MM-dd";public static SimpleDateFormat sdf&＃61;new SimpleDateFormat(date_short);public static String getCountDate(String date,String patton){SimpleDateFormat sdf&＃61;new SimpleDateFormat(patton);Calendar cal &＃61;Calendar.getInstance();if (date!&＃61;null){try {cal.setTime(sdf.parse(date));} catch (ParseException e) {e.printStackTrace();}}return sdf.format(cal.getTime());}public static Date parseDate(String dateStr) throws Exception{return sdf.parse(dateStr);}public static void main(String[] args) {System.out.println(DateFmt.getCountDate("2015-09-08 09:09:08 ", DateFmt.date_long));}
}