[BD] Storm

什么是实时计算

• 离线计算：批处理，代表MapReduce、Spark Core，采集数据Sqoop、Flume
• 实时计算：源源不断，代表Storm等，采集数据Flume
• 框架
  • Apache Storm
  • Spark Streaming：把流式数据转换成离散数据，本质是离线计算
  • JStrom：阿里基于Strom开发
  • Flink

 环境搭建

• 伪分布
  • storm nimbus &
  • storm supervisor &
  • storm ui &

• 全分布

编程案例 WordCount

• 启用Debug，日志查看器，在网页上查看数据
  • "topology.eventlogger.executors": 1
  • /root/training/apache-storm-1.0.3/examples/storm-starter 
  • storm jar storm-starter-topologies-1.0.3.jar org.apache.storm.starter.WordCountTopology MyWC
  • storm logviewer &

WordCountTopology.java

package demo.wc;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.hdfs.bolt.HdfsBolt;
import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;
import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy;
import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy.Units;
import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;
import org.apache.storm.redis.bolt.RedisStoreBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisStoreMapper;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;


//任务的主程序，创建任务：Topology
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"));
        
        //设置任务的第三个Bolt组件，将结果保存到Redis，直接使用Storm提供的BOlt
        //builder.setBolt("wordcount_redis", createRedisBolt()).shuffleGrouping("wordcount_total");
        
        //设置任务的第三个Bolt组件，将结果保存到HDFS（文件），直接使用Storm提供的Bolt
        builder.setBolt("wordcount_hdfs", createHDFSBolt()).shuffleGrouping("wordcount_total");
        
        //设置任务的第三个Bolt组件，将结果保存到HBase中
        //builder.setBolt("wordcount_hbase", new WordCountHBaseBolt()).shuffleGrouping("wordcount_total");
        
        
        //创建一个任务：Topology
        StormTopology topology = builder.createTopology();
        
        //创建一个Config对象，保存配置信息
        Config conf = new Config();
        
        /*
         * 提交Storm的任务有两种方式
         * 1、本地模式
         * 2、集群模式
         */
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("MyWordCount", conf, topology);
        
//        StormSubmitter.submitTopology("MyWordCount", conf, topology);

    }

    private static IRichBolt createHDFSBolt() {
        // 将结果保存到HDFS 文件
        
        HdfsBolt bolt = new HdfsBolt();
        //设置HDFS的相关配置信息
        //HDFS的位置：NameNode的地址
        bolt.withFsUrl("hdfs://192.168.174.111:9000");
        
        //设置保存的HDFS的目录
        bolt.withFileNameFormat(new DefaultFileNameFormat().withPath("/stormdata"));
        
        //保存的是<key value>，设置数据保存到文件的时候，分隔符 |
        //举例：<Beijing,10>   ----> 结果: Beijing|10
        bolt.withRecordFormat(new DelimitedRecordFormat().withFieldDelimiter("|"));
        
        //流式处理，多大的数据生成一个文件?
        //每5M的数据生成一个文件
        bolt.withRotationPolicy(new FileSizeRotationPolicy(5.0f, Units.MB));
        
        //当输出tuple达到了一定大小，就会跟HDFS进行一次同步
        bolt.withSyncPolicy(new CountSyncPolicy(1000));
        
        
        return bolt;
    }

    private static IRichBolt createRedisBolt() {
        //把单词计数是结果保存到Redis
        
        //创建Redis的连接池
        JedisPoolConfig.Builder builder = new JedisPoolConfig.Builder();
        builder.setHost("192.168.174.111");
        builder.setPort(6379);
        JedisPoolConfig poolConfig = builder.build();
        
        //参数：StoreMapper：用于指定存入Redis中的数据格式
        return new RedisStoreBolt(poolConfig, new RedisStoreMapper() {
            
            @Override
            public RedisDataTypeDescription getDataTypeDescription() {
                //定义Redis中的数据类型：WordCount采用什么数据类型？
                //使用Hash集合
                return new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH,
                                                    "wordcount");
            }
            
            @Override
            public String getValueFromTuple(ITuple tuple) {
                // 从上一个Tuple中取出值：频率
                return String.valueOf(tuple.getIntegerByField("total"));
            }
            
            @Override
            public String getKeyFromTuple(ITuple tuple) {
                // 从上一个Tuple中取出key：单词
                return tuple.getStringByField("word");
            }
        });
    }
}

WordCountSpout.java

package demo.wc;

import java.util.Map;
import java.util.Random;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

//第一级组件，作为任务的Spout组件，来采集数据
//模拟一些数据，随机产生数据
public class WordCountSpout extends BaseRichSpout {

    //定义我们要产生的数据
    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];
        
        //把数据发送给下一个组件
        //数据一定要遵循schema的结构
        System.out.println("采集的数据是：" + data);
        this.collector.emit(new Values(data));
    }

    @Override
    public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {
        //相当于Spout初始化方法
        //参数：SpoutOutputCollector collector 相当于是输出流
        this.collector = collector;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declare) {
        // 申明Tuple的格式，是Schema
        declare.declare(new Fields("sentence"));
    }
}

WordCountSplitBolt.java

package demo.wc;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

//第二级组件，是bolt组件，用于单词的拆分
public class WordCountSplitBolt extends BaseRichBolt{

    private OutputCollector collector;
    
    @Override
    public void execute(Tuple tuple) {
        //如何处理上一级组件发来的数据: I love Beijing
        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 declare) {
        //申明Tuple的格式
        declare.declare(new Fields("word","count"));
        
    }
}

WordCountTotalBolt.java

package demo.wc;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

//第三级组件，是bolt组件，用于单词的计数
public class WordCountTotalBolt extends BaseRichBolt {

    private OutputCollector collector;
    
    //定义一个Map集合来保存结果
    private Map<String, Integer> result = new HashMap<>();
    
    @Override
    public void execute(Tuple tuple) {
        // 对每个单词进行计数
        //取出数据
        String word = tuple.getStringByField("word");
        int count = tuple.getIntegerByField("count");
        
        if(result.containsKey(word)){
            //如果包含，进行累加
            int total = result.get(word);
            result.put(word, total+count);
        }else{
            //这个单词第一次出现
            result.put(word, count);
        }
        
        //打印在屏幕上
        System.out.println("统计的结果是: " + result);
        
        //把结果继续发送给下一个bolt组件:  (单词,频率)
        this.collector.emit(new Values(word,result.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"));
    }
}

编程模型

• Topology：Storm中运行的一个实时应用程序
• Stream：数据流向
• Spout：在一个Topology中获取源数据流的组件
• Bolt：接收数据然后执行处理的组件，可级联
• Tuple：一次消息传递的基本单元
• StreamGroup：数据分组策略
  • 随机分组：1-2之间
  • 按字段分组：2-3之间
  • 广播分组

流式计算架构

• Flume：获取数据
• Kafka：临时保存数据
• Storm：计算数据
• Redis：保存数据

 原理分析

• Storm在ZK中保存的数据

• Storm提交任务的过程

 

• Storm内部通信的机制

外部集成

• Redis
  • 添加依赖jar包，在WordCountTopology中编写Bolt组件
  • 创建连接池

• JDBC
• HDFS：storm-hdfs***.jar
• HBase：自己开发一个Bolt组件
• Kafka
• Hive

参考

大数据实时计算框架

https://www.csdn.net/gather_21/MtTacgxsMDI1Mi1ibG9n.html