起风网1.Storm第一个Demo
2.Windows下基于eclipse的Storm应用开发与调试
3.Storm实例+mysql数据库保存
4.Storm原理介绍
5. flume+kafka+storm+mysql 实时架构
1.Storm第一个Demo
Storm运行模式：
- 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解） 运行在本地机器的单一JVM上，这个模式主要用来开发、调试。
- 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式。
写一个HelloWord Storm
我们现在创建这么一个应用，统计文本文件中的单词个数，详细学习过Hadoop的朋友都应该写过。那么我们需要具体创建这样一个Topology，用一个spout负责读取文本文件，用第一个bolt来解析成单词，用第二个bolt来对解析出的单词计数，整体结构如图所示：
写一个可运行的Demo很简单，我们只需要三步：
- 创建一个Spout读取数据
- 创建bolt处理数据
- 创建一个Topology提交到集群
下面我们就写一下，以下代码拷贝到eclipse（依赖的jar包到官网下载即可）即可运行。
1.创建一个Spout作为数据源
Spout作为数据源，它实现了IRichSpout接口，功能是读取一个文本文件并把它的每一行内容发送给bolt。
package storm.demo.spout;import java.io.BufferedReader;import java.io.FileNotFoundException;import java.io.FileReader;import java.util.Map;import backtype.storm.spout.SpoutOutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.IRichSpout;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values;public class WordReader implements IRichSpout {private static final long serialVersionUID = 1L;private SpoutOutputCollector collector;private FileReader fileReader;private boolean completed = false;public boolean isDistributed() {return false;}/** * 这是第一个方法，里面接收了三个参数，第一个是创建Topology时的配置， * 第二个是所有的Topology数据，第三个是用来把Spout的数据发射给bolt * **/@Overridepublic void open(Map conf, TopologyContext context,SpoutOutputCollector collector) {try {//获取创建Topology时指定的要读取的文件路径this.fileReader = new FileReader(conf.get("wordsFile").toString());} catch (FileNotFoundException e) {throw new RuntimeException("Error reading file ["+ conf.get("wordFile") + "]");}//初始化发射器this.collector = collector;}/** * 这是Spout最主要的方法，在这里我们读取文本文件，并把它的每一行发射出去（给bolt） * 这个方法会不断被调用，为了降低它对CPU的消耗，当任务完成时让它sleep一下 * **/@Overridepublic void nextTuple() {if (completed) {try {Thread.sleep(1000);} catch (InterruptedException e) {// Do nothing}return;}String str;// Open the readerBufferedReader reader = new BufferedReader(fileReader);try {// Read all lineswhile ((str = reader.readLine()) != null) {/** * 发射每一行，Values是一个ArrayList的实现 */this.collector.emit(new Values(str), str);}} catch (Exception e) {throw new RuntimeException("Error reading tuple", e);} finally {completed = true;}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("line"));}@Overridepublic void close() {// TODO Auto-generated method stub}@Overridepublic void activate() {// TODO Auto-generated method stub}@Overridepublic void deactivate() {// TODO Auto-generated method stub}@Overridepublic void ack(Object msgId) {System.out.println("OK:" + msgId);}@Overridepublic void fail(Object msgId) {System.out.println("FAIL:" + msgId);}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}}
2.创建两个bolt来处理Spout发射出的数据
Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来，我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。
Bolt中最重要的是execute方法，每当一个tuple传过来时它便会被调用。
第一个bolt：WordNormalizer
package storm.demo.bolt;import java.util.ArrayList;import java.util.List;import java.util.Map;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.IRichBolt;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;public class WordNormalizer implements IRichBolt {private OutputCollector collector;@Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.collector = collector;}/**这是bolt中最重要的方法，每当接收到一个tuple时，此方法便被调用 * 这个方法的作用就是把文本文件中的每一行切分成一个个单词，并把这些单词发射出去（给下一个bolt处理） * **/@Overridepublic void execute(Tuple input) {String sentence = input.getString(0);String[] words = sentence.split(" ");for (String word : words) {word = word.trim();if (!word.isEmpty()) {word = word.toLowerCase();// Emit the wordList a = new ArrayList();a.add(input);collector.emit(a, new Values(word));}}//确认成功处理一个tuplecollector.ack(input);}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}@Overridepublic void cleanup() {// TODO Auto-generated method stub}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}}
第二个bolt:WordCounter
package storm.demo.bolt;import java.util.HashMap;import java.util.Map;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.IRichBolt;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Tuple;public class WordCounter implements IRichBolt {Integer id;String name;Map<String, Integer> counters;private OutputCollector collector;@Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.counters = new HashMap<String, Integer>();this.collector = collector;this.name = context.getThisComponentId();this.id = context.getThisTaskId();}@Overridepublic void execute(Tuple input) {String str = input.getString(0);if (!counters.containsKey(str)) {counters.put(str, 1);} else {Integer c = counters.get(str) + 1;counters.put(str, c);}// 确认成功处理一个tuplecollector.ack(input);}/** * Topology执行完毕的清理工作，比如关闭连接、释放资源等操作都会写在这里 * 因为这只是个Demo，我们用它来打印我们的计数器 * */@Overridepublic void cleanup() {System.out.println("-- Word Counter [" + name + "-" + id + "] --");for (Map.Entry<String, Integer> entry : counters.entrySet()) {System.out.println(entry.getKey() + ": " + entry.getValue());}counters.clear();}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stub}@Overridepublic Map<String, Object> getComponentConfiguration() {// TODO Auto-generated method stubreturn null;}}
3.在main函数中创建一个Topology
在这里我们要创建一个Topology和一个LocalCluster对象，还有一个Config对象做一些配置。
package storm.demo;import storm.demo.bolt.WordCounter;import storm.demo.bolt.WordNormalizer;import storm.demo.spout.WordReader;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.topology.TopologyBuilder;import backtype.storm.tuple.Fields;public class WordCountTopologyMain {public static void main(String[] args) throws InterruptedException {//定义一个TopologyTopologyBuilder builder = new TopologyBuilder();builder.setSpout("word-reader",new WordReader());builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader");builder.setBolt("word-counter", new WordCounter(),2).fieldsGrouping("word-normalizer", new Fields("word"));//配置Config conf = new Config();conf.put("wordsFile", "d:/text.txt");conf.setDebug(false);//提交Topologyconf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);//创建一个本地模式clusterLocalCluster cluster = new LocalCluster();cluster.submitTopology("Getting-Started-Toplogie", conf,builder.createTopology());Thread.sleep(1000);cluster.shutdown();}}
运行这个函数我们即可看到后台打印出来的单词个数。
3.Storm实例+mysql数据库保存
package com.qing.storm.Spout;import java.io.BufferedReader;import java.io.FileInputStream;import java.io.InputStreamReader;import java.util.Map;import backtype.storm.spout.SpoutOutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseRichSpout;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values;@SuppressWarnings("serial")public class ReadLogSpout extends BaseRichSpout {private SpoutOutputCollector collector;FileInputStream fis;InputStreamReader isr;BufferedReader br;@Overridepublic void nextTuple() {// TODO Auto-generated method stubString str = "";try {while ((str = this.br.readLine()) != null) {this.collector.emit(new Values(str));Thread.sleep(100);}} catch (Exception e) {e.printStackTrace();}}@SuppressWarnings("rawtypes")@Overridepublic void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {// TODO Auto-generated method stubthis.collector = collector;String file = "/opt/apache-storm-0.9.3/bin/domain.log";try{this.fis = new FileInputStream(file);this.isr = new InputStreamReader(fis);this.br = new BufferedReader(isr);} catch (Exception e) {e.printStackTrace();}}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stubdeclarer.declare(new Fields("str"));}}///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////package com.qing.storm.Bolt;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseBasicBolt;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;@SuppressWarnings("serial")public class SplitBolt extends BaseBasicBolt{@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stubdeclarer.declare(new Fields("word"));}@Overridepublic void execute(Tuple tuple, BasicOutputCollector collector) {// TODO Auto-generated method stubString sentence = tuple.getString(0);for(String word: sentence.split(" ")){collector.emit(new Values(word));}}}///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////package com.qing.storm.Bolt;import java.util.HashMap;import java.util.Map;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseBasicBolt;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;@SuppressWarnings("serial")public class WordCountBolt extends BaseBasicBolt{Map<String, Integer> counts = new HashMap<String, Integer>();public void execute(Tuple tuple, BasicOutputCollector collector) {// TODO Auto-generated method stubString word = tuple.getString(0);Integer count = counts.get(word);if(count == null){count = 0;}count++;counts.put(word, count);collector.emit(new Values(word,count));}@Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {// TODO Auto-generated method stubdeclarer.declare(new Fields("word", "count"));}}///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////package com.qing.storm.Bolt;import java.sql.Connection;import java.sql.DriverManager;import java.sql.SQLException;import java.sql.Statement;import java.util.Map;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseRichBolt;import backtype.storm.tuple.Tuple;@SuppressWarnings("serial")public class MysqlBolt extends BaseRichBolt{private OutputCollector collector;Connection conn = null;String from = "word_count"; //表名private String word;private int num;@Overridepublic void declareOutputFields(OutputFieldsDeclarer arg0) {// TODO Auto-generated method stub}@Overridepublic void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector) {// TODO Auto-generated method stubthis.collector = collector;try {LinkDB();} catch (InstantiationException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (IllegalAccessException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();}}private void LinkDB() throws InstantiationException, IllegalAccessException, SQLException {// TODO Auto-generated method stubString host_port = "127.0.0.1:3306";String database = "storm_test";String username = "root";String password = "root";String url = "jdbc:mysql://" + host_port + "/" + database;try {Class.forName("com.mysql.jdbc.Driver");conn = DriverManager.getConnection(url, username, password);} catch (ClassNotFoundException e) {// TODO Auto-generated catch blocke.printStackTrace();}}@Overridepublic void execute(Tuple tuple) {// TODO Auto-generated method stubString word= tuple.getString(0);int num = tuple.getInteger(1);InsertDB(word, num);}private void InsertDB(String word, int num) {// TODO Auto-generated method stubthis.word = word;this.num = num;String sql = "replace into " + this.from+ "(word, num) values ('" +word+"',"+num+ ")";try { Statementstatement = conn.createStatement();statement.executeUpdate(sql);} catch (SQLException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////package com.qing.storm.Topology;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.StormSubmitter;import backtype.storm.generated.AlreadyAliveException;import backtype.storm.generated.InvalidTopologyException;import backtype.storm.topology.TopologyBuilder;import backtype.storm.tuple.Fields;import com.qing.storm.Bolt.MysqlBolt;import com.qing.storm.Bolt.SplitBolt;import com.qing.storm.Bolt.WordCountBolt;import com.qing.storm.Spout.ReadLogSpout;public class DB_Topology {public static void main(String[] args){ try {TopologyBuilder builder = new TopologyBuilder();builder.setSpout("spout", new ReadLogSpout(), 5);builder.setBolt("split", new SplitBolt(), 8).shuffleGrouping("spout");builder.setBolt("count", new WordCountBolt(), 10).fieldsGrouping("split", new Fields("word"));builder.setBolt("Mysql", new MysqlBolt(),10).fieldsGrouping("count", new Fields("word","count"));Config conf = new Config();conf.setDebug(true); //if (args != null && args.length > 0) { /*设置该topology在storm集群中要抢占的资源slot数，一个slot对应这supervisor节点上的以个worker进程 如果你分配的spot数超过了你的物理节点所拥有的worker数目的话，有可能提交不成功，加入你的集群上面已经有了 一些topology而现在还剩下2个worker资源，如果你在代码里分配4个给你的topology的话，那么这个topology可以提交 但是提交以后你会发现并没有运行。 而当你kill掉一些topology后释放了一些slot后你的这个topology就会恢复正常运行。 */ //conf.setNumWorkers(1); if (args != null && args.length > 0) { conf.setNumWorkers(1); StormSubmitter.submitTopology(args[0], conf, builder.createTopology()); }} catch (AlreadyAliveException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (InvalidTopologyException e) {// TODO Auto-generated catch blocke.printStackTrace();} }//conf.setMaxTaskParallelism(1);//LocalCluster cluster = new LocalCluster();//cluster.submitTopology("word-count", conf, builder.createTopology());//try {//Thread.sleep(1000000);//} catch (InterruptedException e) {//// TODO Auto-generated catch block//e.printStackTrace();//}//cluster.shutdown();}
4.Storm原理介绍
Storm是使用Clojure语言开发，但是可以在Storm中使用任何语言编写应用程序，所需的只是一个连接到Storm 的架构的适配器。已存在针对 Scala、JRuby、Perl 和 PHP 的适配器，但是还有支持流式传输到 Storm 拓扑结构中的结构化查询语言适配器。
Storm 的关键属性
Storm 实现的一些特征决定了它的性能和可靠性的。Storm 使用 ZeroMQ 传送消息，这就消除了中间的排队过程，使得消息能够直接在任务自身之间流动。在消息的背后，是一种用于序列化和反序列化 Storm 的原语类型的自动化且高效的机制。
Storm 的一个最有趣的地方是它注重容错和管理。Storm 实现了有保障的消息处理，所以每个元组都会通过该拓扑结构进行全面处理；如果发现一个元组还未处理会自动从喷嘴处重放。Storm 还实现了任务级的故障检测，在一个任务发生故障时，消息会自动重新分配以快速重新开始处理。Storm 包含比 Hadoop 更智能的处理管理，流程会由监管员来进行管理，以确保资源得到充分使用。
Storm 实现了一种数据流模型，其中数据持续地流经一个转换实体网络（参见 图 1）。一个数据流的抽象称为一个流，这是一个无限的元组序列。元组就像一种使用一些附加的序列化代码来表示标准数据类型（比如整数、浮点和字节数组）或用户定义类型的结构。每个流由一个惟一 ID 定义，这个 ID 可用于构建数据源和接收器 (sink) 的拓扑结构。流起源于喷嘴，喷嘴将数据从外部来源流入 Storm 拓扑结构中。
接收器（或提供转换的实体）称为螺栓。螺栓实现了一个流上的单一转换和一个 Storm 拓扑结构中的所有处理。螺栓既可实现 MapReduce 之类的传统功能，也可实现更复杂的操作（单步功能），比如过滤、聚合或与数据库等外部实体通信。典型的 Storm 拓扑结构会实现多个转换，因此需要多个具有独立元组流的螺栓。喷嘴和螺栓都实现为 Linux® 系统中的一个或多个任务。
使用 Storm 为词频轻松地实现 MapReduce 功能。如 图 2 中所示，喷嘴生成文本数据流，螺栓实现 Map 功能（令牌化一个流的各个单词）。来自 “map” 螺栓的流然后流入一个实现 Reduce 功能的螺栓中（以将单词聚合到总数中）。
请注意，螺栓可将数据传输到多个螺栓，也可接受来自多个来源的数据。Storm 拥有流分组 的概念，流分组实现了混排 (shuffling)（随机但均等地将元组分发到螺栓）或字段分组（根据流的字段进行流分区）。还存在其他流分组，包括生成者使用自己的内部逻辑路由元组的能力。
但是，Storm 架构中一个最有趣的特性是有保障的消息处理。Storm 可保证一个喷嘴发射出的每个元组都会处理；如果它在超时时间内没有处理，Storm 会从该喷嘴重放该元组。此功能需要一些聪明的技巧来在拓扑结构中跟踪元素，也是 Storm 的重要的附加价值之一。
除了支持可靠的消息传送外，Storm 还使用 ZeroMQ 最大化消息传送性能（删除中间排队，实现消息在任务间的直接传送）。ZeroMQ 合并了拥塞检测并调整了它的通信，以优化可用的带宽。
5. flume+kafka+storm+mysql 实时架构
flume的架构图：
kafka的架构图：
storm的架构图：
flume + kafka + storm +mysql的数据流架构图：
下面介绍一下kafka到storm的配置：
其实这些都是通过java代码实现的，这里用到了 KafkaSpout类，RDBMSDumperBolt类（以后这些可以作为工具类打包上传到集群中）
storm作业中，我们写了一个KafkaStormRdbms类，作业具体配置如下：
首先设置连接mysql的参数
ArrayList<String> columnNames = new ArrayList<String>(); ArrayList<String> columnTypes = new ArrayList<String>(); String tableName = "stormTestTable_01"; // Note: if the rdbms table need not to have a primary key, set the variable 'primaryKey' to 'N/A' // else set its value to the name of the tuple field which is to be treated as primary key String primaryKey = "N/A"; String rdbmsUrl = "jdbc:mysql://$hostname:3306/fuqingwuDB" ; String rdbmsUserName = "fuqingwu"; String rdbmsPassword = "password"; //add the column names and the respective types in the two arraylists columnNames.add("word"); //add the types columnTypes.add("varchar (100)");
配置 KafkaSpout 及 Topology：
TopologyBuilder builder = new TopologyBuilder();List<String> hosts = new ArrayList<String>(); hosts.add("hadoop01"); SpoutConfig spoutConf = SpoutConfig.fromHostStrings(hosts, 1, "flume_kafka", "/root", "id"); spoutConf.scheme = new StringScheme(); spoutConf.forceStartOffsetTime(-2); spoutConf.zkServers = new ArrayList<String>() {{ add("hadoop01"); }}; spoutConf.zkPort = 2181; //set the spout for the topologybuilder.setSpout("spout", new KafkaSpout(spoutConf), 1);//dump the stream data into rdbms tableRDBMSDumperBolt dumperBolt = new RDBMSDumperBolt(primaryKey, tableName, columnNames, columnTypes, rdbmsUrl, rdbmsUserName, rdbmsPassword);builder.setBolt("dumperBolt",dumperBolt, 1).shuffleGrouping("spout");
原文连接：http://blog.csdn.net/baiyangfu_love/article/details/8096088
GitHub：https://github.com/baniuyao/flume-kafka
这个框架用的组件基本都是最新稳定版本，flume-ng1.4+kafka0.8+storm0.9+mysql架构设计：
1）.数据采集
负责从各节点上实时采集数据，选用cloudera的flume来实现
2）.数据接入
由于采集数据的速度和数据处理的速度不一定同步，因此添加一个消息中间件来作为缓冲，选用apache的kafka
3）.流式计算
对采集到的数据进行实时分析，选用apache的storm
4）.数据输出
对分析后的结果持久化，暂定用mysql
参考：http://blog.csdn.net/mylittlered/article/details/20810265
http://www.blogchong.com/post/storm_data_Platform.html
评论前必须登录！
立即登录