大数据实时计算框架：storm

(一)什么是实时计算？跟离线计算的区别？常见的实时计算框架？

1.什么是实时计算？流式计算
  举例：自来水厂处理水的过程（图）
        特点：源源不断
	任务类型：采集数据-->Spout任务
	          处理数据-->bolt任务

2.跟离线计算的区别
	(1)离线计算：MapReduce、spark core
	   采集数据：SQOOP，flume
	   强调是批处理
	(2)实时计算：storm
	   采集数据：flume
	   强调是源源不断

3.常见的实时计算框架
 (1)Apache storm
 (2)Spark Streaming
 (3)JStorm
 (4)Flink

(二)storm架构

主从
    nimbus+supervisor

(三)伪分布式Storm安装

解压
设置环境变量
    STORM_HOME=/root/training/apache-storm-1.3.0
    export STORM_HOME
    PATH=$STORM_HOME/bin:$PATH
    export PATH
核心配置文件：conf/storm.yaml
    storm.zookeeper.servers:

   - "bigdata111"
         nimbus.seeds:["bigdata111"]
配置Supervisor上的slot的个数（端口号）
     supervisor.slots.ports:
       - 6700
         - 6701
          - 6702
          - 6703
在nimbus配置一个目录，保存任务和元信息 在nimbus在创建一个tmp目录
            	storm.local.dir: "/root/training/apache-storm-1.0.3/tmp"
启动：
     - 主节点：storm nimbus &
      - 从节点：storm supervisor &
            UI：storm ui & //地址：192.168.11.111:8080

(四)全分布式Storm

启动zookeeper
   		解压
    	设置环境变量
             STORM_HOME=/root/training/apache-storm-1.3.0
             export STORM_HOME
             PATH=$STORM_HOME/bin:$PATH
             export PATH
核心配置文件：conf/storm.yaml
            storm.zookeeper.servers:
          - "hadoop112"
          - "hadoop113"
          - "hadoop114"
            nimbus.seeds:["hadoop112"]
配置Supervisor上的slot的个数（端口号）
            supervisor.slots.ports:
          - 6700
          - 6701
          - 6702
          - 6703
在nimbus配置一个目录，保存任务和元信息 在nimbus在创建一个tmp目录
            storm.local.dir: "/root/training/apache-storm-1.0.3/tmp"
复制storm到hadoop113/hadoop114节点
启动：
    -  主节点：storm nimbus &
    - 从节点：storm supervisor &
            UI：storm ui & //地址：192.168.11.112:8080

(五)storm HA

            
步骤和全分布式一致
       nimbus.seeds:["hadoop112","hadoop113"]
 启动：
      - 主节点：storm nimbus &
      - 备用主节点：storm nimbus &
      - 从节点：storm supervisor &
            UI：storm ui & //地址：192.168.11.112:808(主节点上)
 

(六)storm demo

启用Debug
核心配置文件：conf/storm.yaml
         "topology.eventlogger.executors": 1
启动节点后启动日志查看器 storm logviewer &
      案例： wordcount程序
提交任务命令格式：storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】
在storm/example/storm-starter目录下
       storm jar storm-starter-topologies-1.0.3.jar org.apache.storm.starter.WordCountTopology MyWordCount //别名
	//将jar包上传到nimbus的tmp目录下

(七)开发Wordcount程序

第一级：WordCountSpout 采集数据组件
第二级：WordCountSplitBolt 单词拆分组件
            							注意：组件之间传递都是Tuple Tuple=schema+数据
数据分组的策略（数据到底交给哪个下级组件处理）
         (1)随机分组
         (2)按字段分组 同MapReduce
         (3)广播分组 （所有下级组件都能收到这条数据）
第三级：WordCountTotalBolt 单词计数组件
           							 注意：第二级和三级之间用按字段分组策略

(1)开发WordCountSpout组件

             public class WordCountSpout extends BaseRichSpout {
    
            private static final long serialVersionUID = 1L;
            //定义要产生的数据
            private String[] datas = {
    "I love Beijing","I love China","Beijing is the capital of China"};
            //定义一个变量保存输出流
            private SpoutOutputCollector collector;
            @Override
            public void nextTuple() {
    
            //每隔2秒采集一次
            Utils.sleep(2000);
            // 又storm框架调用，用于如何接收数据
            //产生3以内的随机数
            int random = (new Random()).nextInt(3);
            String data = datas[random];
            //发送给下一级组件
            System.out.println("采集的数数是:"+data);
            this.collector.emit(new Values(data));

	}
	@Override
	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
    
		// 相当于初始化方法
		// SpoutOutputCollector collector : 输出流
		this.collector = collector;
		
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
    
		// 声明tuple格式
		declare.declare(new Fields("sentence"));
	}

}

(2)开发WordCountSplitBolt组件

 public class WordCountSplitBolt extends BaseRichBolt {
    
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	@Override
	public void execute(Tuple tuple) {
    
		// 如何处理上一级组件发来的数据
		String data = tuple.getStringByField("sentence");
		//分词
		String[] words = data.split(" ");
		//输出
		for(String w : words) {
    
			collector.emit(new Values(w,1));
		}
	}
	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
    
		// bolt进行初始化
		this.collector = collector;
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer d) {
    
		// 申明下级tuple的格式
		d.declare(new Fields("word","count"));
	}
}

(3)开发WordCountTotalBolt组件

 public class WordCountTotalBolt extends BaseRichBolt {
    
	private static final long serialVersionUID = 1L;	
	private OutputCollector collector;
	private Map<String, Integer> resut = new HashMap<>();
	@Override
	public void execute(Tuple tuple) {
    
		// 对每个单词进行计数
		//取数据
		String word = tuple.getStringByField("word");
		int count = tuple.getIntegerByField("count");
		if(resut.containsKey(word)) {
    
			int total = resut.get(word);
			resut.put(word, total+count);
		}
		else {
    
			resut.put(word, count);
		}
		System.out.println("统计结果是:"+resut);
		//结果发动到下一级组件
		this.collector.emit(new Values(word,resut.get(word)));
	}
	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
    
		// TODO Auto-generated method stub
		this.collector = collector;
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
    
		// TODO Auto-generated method stub
		declare.declare(new Fields("word","total"));
	}
}

(4)开发主程序WordCountTopology

 public class WordCountTopology {
    
	public static void main(String[] args) throws Exception{
    
		TopologyBuilder builder = new TopologyBuilder();
		//设置spout组件
		builder.setSpout("wordcount_spout", new WordCountSpout());	
		//设置单词拆分bolt组件
		builder.setBolt("wordcount_split", new WordCountSplitBolt()).
			shuffleGrouping("wordcount_spout");		
		//设置任务的单词计数的bolt组件，是按字段分组
		builder.setBolt("wordcount_total",new WordCountTotalBolt()).
			fieldsGrouping("wordcount_split", new Fields("word"));		
		//创建一个任务
		StormTopology topology = builder.createTopology();	
		//创建Configure对象
		Config conf = new Config();		
		//提交Storm的任务两种方式
		//1 本地模式
//		LocalCluster cluster = new LocalCluster();
//		cluster.submitTopology("Mywordcount", conf, topology);		
		//2.集群模式
		StormSubmitter.submitTopology("Mywordcount", conf, topology);
	}
}

(5)提交到storm集群

 storm jar storm_wordcount.jar com.hadoop.storm.WordCountTopology
 (八)storm原理分析
 (1).Storm在ZK中保存的数据
 (2).storm任务提交的过程
 (3).Storm内部通信机制：Worker之间通信的基本原理
 (九)Storm集成
 (1)集成HBASE
 /**

 * 建表 ： create 'result','info'

 * @author cheng
   */
   public class WordCountHBaseBolt extends BaseRichBolt {
    
   private static final long serialVersionUID = 1L;
   private Table client = null;
   @Override
   public void execute(Tuple tuple) {
    
   	// 把上一个组件发来的数据存入HBase
   	String word = tuple.getStringByField("word");
   	int total = tuple.getIntegerByField("total");
   	//构造Put
   	Put put = new Put(Bytes.toBytes(word));
   	put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("word"), Bytes.toBytes(word));
   	put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("total"), Bytes.toBytes(String.valueOf(total)));
   	//插入HBase
   	try {
    
   		client.put(put);
   	} catch (Exception e) {
    
   		// TODO Auto-generated catch block
   		e.printStackTrace();
   	}
   }

   @Override
   public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
    
   	try {
    
   		// TODO Auto-generated method stub
   		Configuration conf = new Configuration();
   		conf.set("hbase.zookeeper.quorum", "192.168.11.111");
   		//获取连接对象
   		Connection conn = ConnectionFactory.createConnection(conf);
   		client = conn.getTable(TableName.valueOf("result"));
   	}
   	catch (Exception e) {
    
   		e.printStackTrace();
   	}	
   }
   @Override
   public void declareOutputFields(OutputFieldsDeclarer arg0) {
    
   	// TODO Auto-generated method stub
   }
   }
   在Topology中添加HBaseBolt
   builder.setBolt("wordcount_hbase", new WordCountHBaseBolt()).
   		shuffleGrouping("wordcount_total");