Storm 工作原理

Storm 工作原理


Storm简介
1.Storm是一套分布式的、可靠的，可容错的用于处理流式数据的系统。
2.Storm也是基于C/S架构来进行工作的,C负责将数据处理的方式的jar(Topology)发送给S,S解析C发送过来的jar(Topology)，并按一定规则jar变成多个Task((Spout/Bolt)),生成相关的进程和线程运行里面的Task。


相关述语说明:
1.Topology(拓扑):storm中运行的一个实时应用程序(Storm的一个任务单元)，因为各个组件间的消息流动形成逻辑上的一个拓扑结构(所以叫Topology)。Topolog是一组由Spouts（数据源）和Bolts（数据操作）通过Stream Groupings进行连接组成的图。
2.tuple（元组）：一次消息传递的基本单元。本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个value list.
3.Stream：以tuple为单位组成的一条有向无界的数据流。（就是tuple在各个组件中流动时的描述）
4.Spout组件:就是一个继承了某个基类的类,里面有类的方法进行相关的操作,用于获取数据，并传递数据到Bolt。
5.Bolt组件:就是一个继承了某个基类的类,里面有类的方法进行相关的操作,用于对Spout组件发送过来的数据进行处理。
6.Worker进程,用于运行Topology子集(可能Topology的不同组件(Spout/Bolt)会放在不同的Worker进程来运行)的进程。
7.executor线程，为Worker进程中的一个线程,executor可能会同时运行多个组件(Spout/Bolt),当然同一个executor运行的组件类型是一样的。
8.Task,任务，就是组件(Spout/Bolt),一般是一个executor线程运行一个Task
9.Nimbus进程，控制节点(Nimbus节点)，主结点运行一个叫做Nimbus的守护进程，它负责在集群内分发代码，为每个工作结点指派任务和监控失败的任务。
10.Supervisor进程，工作节点(Supervisor节点),工作结点运行一个叫做Supervisor的守护进程，每个工作节点都是topology中一个子集的实现。
11.zookeeper，集群协调软件（C/S），是完成nimbus和supervisor之间协调的服务。
12.storm UI,只提供对topology的监控和统计。


架构图：



topology工作原理
1.Storm集群中有两种节点，一种是控制节点(Nimbus节点)，另一种是工作节点(Supervisor节点)。
2.所有Topology任务的 提交必须在Storm客户端节点上进行(需要配置 storm.yaml文件)，由Nimbus节点分配给其他Supervisor节点进行处理。
3.Nimbus节点首先将提交的Topology进行分片（Spout/Bolt），分成一个个的Task，并将Task和Supervisor相关的信息提交到 zookeeper集群上。
4.Supervisor会去zookeeper集群上认领自己的Task，通知自己的Worker进程进行Task的处理。


topology工作流程
1.提交Topology后,Storm会把代码首先存放到Nimbus节点的inbox目录下,之后，会把当前Storm运行的配置生成一个 stormconf.ser文件放到Nimbus节点的stormdist目录中，在此目录中同时还有序列化之后的Topology代码文件
2.在设定Topology所关联的Spouts和Bolts时，可以同时设置当前Spout和Bolt的executor数目和task数目，默认情况下，一个Topology的task的总和是和executor的总和一致的。之后，系统根据worker(Topology的worker配置参数)的数目，尽量平均的分配这些task的执行。worker在哪个supervisor节点上运行是由storm（随机申请到可用的就OK）本身决定的。
3.Storm看一下那些Worker进程可用，就申请worker(Topology的worker配置参数)的数目给这个Topology。
4.Storm尽量平均的分配这些task到worker。
5.任务分配好之后，Nimbus节点会将任务的信息提交到zookeeper集群，同时在zookeeper集群中会有workerbeats节点，这里存储了当前Topology的所有worker进程的心跳信息。
6.Supervisor 节点会不断的轮询zookeeper集群，在zookeeper的assignments节点中保存了所有Topology的任务分配信息、代码存储目 录、任务之间的关联关系等，Supervisor通过轮询此节点的内容，来领取自己的任务，启动worker进程运行。
7.一个Topology运行之后，就会不断的通过Spouts来发送Stream流，通过Bolts来不断的处理接收到的Stream流，Stream流是无界的。
8.最后一步会不间断的执行，除非手动结束Topology。


Spout组件
1.Spout是Stream的消息产生源， Spout组件的实现可以通过继承BaseRichSpout类或者其他Spout类来完成，也可以通过实现IRichSpout接口来实现.
2.nextTuple、ack 和fail 都在spout任务的同一个线程中被循环调用。当没有元组的发射时，应该让nextTuple睡眠一个很短的时间（如一毫秒），以免浪费太多的CPU。
3.继承了BaseRichSpout后，不用实现close、 activate、 deactivate、 ack、 fail 和 getComponentConfiguration 方法，只关心最基本核心的部分。(帮你实现了一个默认的)
4.实现一个Spout，可以直接实现接口IRichSpout，如果不想写多余的代码，可以直接继承BaseRichSpout

public interface ISpout extends Serializable {
  void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
  void close();
  void nextTuple();
  void ack(Object msgId);
  void fail(Object msgId);
}

open()方法
1.初始化方法

close()方法
1.在该spout将要关闭时调用。
2.但是不保证其一定被调用，因为在集群中supervisor节点，可以使用kill -9来杀死worker进程。
3.只有当Storm是在本地模式下运行，如果是发送停止命令，可以保证close的执行

ack(Object msgId)方法
1.成功处理tuple时回调的方法,通常情况下，此方法的实现是将消息队列中的消息移除，防止消息重放

fail(Object msgId)方法
1.处理tuple失败时回调的方法，通常情况下，此方法的实现是将消息放回消息队列中然后在稍后时间里重放

nextTuple()方法
1.这是Spout类中最重要的一个方法。
2.发射一个Tuple到Topology都是通过这个方法来实现的。
3.调用此方法时，让spout发出元组（tuple）到输出器（ouput collector）。
4.这个方法会不断被调用，所以在没有tuple要处理时最好睡眠一会，让出CPU。


Bolt组件
1.Bolt类接收由Spout或者其他上游Bolt类发来的Tuple，对其进行处理。
2.Bolt组件的实现可以通过继承BasicRichBolt类或者IRichBolt接口等来完成
3.实现一个Bolt，可以实现IRichBolt接口或继承BaseRichBolt，如果不想自己处理结果反馈，可以实现 IBasicBolt接口或继承BaseBasicBolt，它实际上相当于自动实现了collector.emit.ack(inputTuple)

prepare方法
1.此方法和Spout中的open方法类似，在集群中一个worker中的task初始化时调用。 它提供了bolt执行的环境

declareOutputFields方法
1.用于声明当前Bolt发送的Tuple中包含的字段(field)，和Spout中类似，就是用new Values(word1,word2)放入到Tuple中时field1->word1,field2->word2的意思，（不用你每次都写MAP中的key）

cleanup方法
1.同ISpout的close方法，在关闭前调用。同样不保证其一定执行。

execute方法
1.这是Bolt中最关键的一个方法，对于Tuple的处理都可以放到此方法中进行。
2.execute接受一个 tuple进行处理，并用prepare方法传入的OutputCollector的ack方法（表示成功）或fail（表示失败）来反馈处理结果。
3.Storm提供了IBasicBolt接口，其目的就是实现该接口的Bolt不用在代码中提供反馈结果了，Storm内部会自动反馈成功。如果你确实要反馈失败，可以抛出FailedException
4.当有Tuple到达时会调用这个接口。


可靠的bolts和不可靠的bolts
1.每个节点都会调用ack(tuple)或fail(tuple)，Storm因此知道一条消息在这个bolt是否失败了，并通知那个/那些制造了这些消息的spout(s)。

class SplitSentence implenents IRichBolt {
    private OutputCollector collector;

    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    public void execute(Tuple tuple) {
        String sentence = tuple.getString(0);
        for(String word : sentence.split(" ")) {
            collector.emit(tuple, new Values(word));
        }
        collector.ack(tuple);
    }

    public void cleanup(){}

    public void declareOutputFields(OutputFieldsDeclarer declarer){
        declar.declare(new Fields("word"));
    }
}

2.如果一个元组没有在设置的超时时间内完成对消息树的处理，就认为这个元组处理失败。默认超时时间为30秒。你可以通过修改Config.TOPOLOGY_MESSAGE_TIMEOUT修改拓扑的超时时间。
3.你处理的每条消息要么是确认的（译者注：collector.ack()）要么是失败的（译者注：collector.fail()）。Storm使用内存跟踪每个元组，所以如果你不调用这两个方法，该任务最终将耗尽内存。
4.一个bolt可以使用emit(streamId, tuple)把元组分发到多个流，其中参数streamId是一个用来标识流的字符串。然后，你可以在TopologyBuilder决定由哪个流订阅它。(多数据流)
5.extends BaseBasicBolt,会在执行execute方法之后自动调用ack方法。(内部实现了，不用你实现)

class SplitSentence extends BaseBasicBolt {
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String sentence = tuple.getString(0);
        for(String word : sentence.split(" ")) {
            collector.emit(new Values(word));
        }
}

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }
}

6.ack方法（表示成功）或fail（表示失败）,只是为实现可靠提供条件，要实现可靠性还要你在spout进行正确和失败的逻辑处理来实现。


TopologyBuilder(Topology定义类)
1.TopologyBuilder将用来创建拓扑，它决定Storm如何安排各节点，以及它们交换数据的方式。
2.在spout和bolts之间通过shuffleGrouping方法连接。这种分组方式决定了Storm会以随机分配方式从源节点向目标节点发送消息。(其中一种方式，就是消息发送到bolt的方式（因为可能会运行多个bolt）)
3.增加组件的方法说明：

TopologyBuilder.setSpout(String id, IRichSpout spout, Number parallelism_hint)
id：这个组件的ID
spout:我们定义好的spout组件
parallelism_hint:并行数，就是这个spout组件并列运行多少个(多任务)(默认1个)

TopologyBuilder.setBolt(String id, IBasicBolt bolt, Number parallelism_hint)
id：这个组件的ID
bolt:我们定义好的Bolt组件
parallelism_hint:并行数，就是这个bolt组件并列运行多少个(多任务)(默认1个)
setBolt返回一个InputDeclarer对象，用于定义Bolt的输入。
builder.setBolt("exclaim2", new ExclamationBolt(), 5).shuffleGrouping("words").shuffleGrouping("exclaim1");//指定Bolt的多个来源

4.Bolt还要指定应该接收那个流作为输入，以何种流分组方式进行流分发(就是按何种方式对tuple进行分送到不同的Bolt(这些Bolt是同一个处理Bolt的不同任务实例))


Config
1.就是一个MAP，你可以增加你需要的参数进去，在spout的open中可以得到这个引用
2.也有一些是定义好的，如：TOPOLOGY_MAX_SPOUT_PENDING(默认1个),缓存spout发送出去的tuple,当缓存中的tuple多于这个值时spout就挂起，等待Bolt消费到小于这个值时再运行。（这个属性只对可靠消息处理有用）

常见的配置
可以为每个拓扑设置大量的配置。一个可以设置的所有配置的列表可以在这个类（backtype.storm.Config）找到。那些前缀为TOPOLOGY的属性可以被特定拓扑所覆盖，其他的是集群配置，不能被覆盖。下面是一些常见的拓扑设置。
（1）Config.TOPOLOGY_WORKERS
这个设置执行topology的工作进程的数量。例如，如果你将这个参数设置为25，则将会有25个Java进程跨集群执行所有的任务。如果你有一个跨拓扑中的所有组件的组合150并行度，每个工作进程将有6个任务作为线程运行。
（2）Config.TOPOLOGY_ACKERS
这是设置任务的数量，该任务将跟踪元组树，当Spout元组已经完全处理时进行检测。Acker是Storm的可靠性模型不可或缺的一部分，你可以在2.5小节“可靠性机制——保证消息处理”阅读到关于它们的更多信息。
（3）Config.TOPOLOGY_MAX_SPOUT_PENDING
这是设置一次可以在Spout任务等待Spout元组的最大数量。等待意味着元组还尚未确认（acked）或失败（failed）。强烈推荐设置这个配置项防止队列溢出。
（4）Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS
这是一个Spout元组在它被认为是失败前必须完全完成的最大超时时间。这个值默认为30秒，这对于大多数拓扑来说是足够的。关于Storm的可靠性模型如何工作可查阅“可靠性机制——保证消息处理”小节以获得更多信息。
（5）Config.TOPOLOGY_SERIALIZATIONS
可以使用这个配置注册更多的序列化器，这样就可以在元组里面使用自定义类型。

Stream Groupings（流分组方式）
1.Stream Grouping定义了一个流在Bolt任务间该如何被切分,说白了就是当有多个处理相同任务的Bolt任务（并行的Bolt）时，将按什么方式分配消息到这些并行的Bolt中的去处理。
2.torm提供的6个Stream Grouping类型：
(1).随机分组（Shuffle grouping）：随机分发tuple到并行的Bolt的任务，保证每个任务获得相等数量的tuple。
(2).字段分组（Fields grouping）：根据指定字段(key)分割数据流，并分组。例如，根据“user-id”字段，相同“user-id”的元组总是分发到同一个任务，不同“user-id”的元组可能分发到不同的任务。
    (a).按那个字段里的值进行分组。不是这个字段(key)的名来分组。
    (b).分组后，字段里值相同的都会发送到同一个Bolt的实例中去处理（并行的Bolt）。
(3).全部分组（All grouping）：tuple被复制到bolt的所有任务。这种类型需要谨慎使用。
(4).全局分组（Global grouping）：全部流都分配到bolt的同一个任务。明确地说，是分配给ID最小的那个task。
(5).无分组（None grouping）：你不需要关心流是如何分组。目前，无分组等效于随机分组。但最终，Storm将把无分组的Bolts放到Bolts或Spouts订阅它们的同一线程去执行（如果可能）。
(6).直接分组（Direct grouping）：这是一个特别的分组类型。元组生产者决定tuple由哪个元组处理者任务接收。

public interface InputDeclarer<T extends InputDeclarer> {
	// 字段分组
	public T fieldsGrouping(String componentId, Fields fields);
	public T fieldsGrouping(String componentId, String streamId, Fields fields);

	// 全局分组
	public T globalGrouping(String componentId);
	public T globalGrouping(String componentId, String streamId);

	// 随机分组
	public T shuffleGrouping(String componentId);
	public T shuffleGrouping(String componentId, String streamId);

	// 本地或者随机分组
	public T localOrShuffleGrouping(String componentId);
	public T localOrShuffleGrouping(String componentId, String streamId);

	// 无分组
	public T noneGrouping(String componentId);
	public T noneGrouping(String componentId, String streamId);

	// 广播分组
	public T allGrouping(String componentId);
	public T allGrouping(String componentId, String streamId);

	// 直接分组
	public T directGrouping(String componentId);
	public T directGrouping(String componentId, String streamId);

	// 自定义分组
	public T customGrouping(String componentId, CustomStreamGrouping grouping);
	public T customGrouping(String componentId, String streamId, CustomStreamGrouping grouping);
	public T grouping(GlobalStreamId id, Grouping grouping);
}

Topology运行方式,Storm有两种运行方式
本地运行的提交方式

LocalCluster cluster = new LocalCluster();
cluster.submitTopology(TOPOLOGY_NAME, conf, builder.createTopology());
Thread.sleep(2000);
cluster.shutdown();

分布式提交方式

StormSubmitter.submitTopology（TOPOLOGY_NAME, conf, builder.createTopology());

注意：
在Storm代码编写完成之后，需要打包成jar包放到Nimbus中运行，打包的时候，不需要把依赖的jar都打迚去，否则如果把依赖的 storm.jar包打进去的话，运行时会出现重复的配置文件错误导致Topology无法运行。因为Topology运行之前，会加载本地的 storm.yaml 配置文件。
运行的命令如下： storm jar StormTopology.jar mainclass [args]


例子：

pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">

	<modelVersion>4.0.0</modelVersion>
	<groupId>storm.book</groupId>
	<artifactId>Getting-Started</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<build>
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>2.3.2</version>
				<configuration>
					<source>1.7</source>
					<target>1.7</target>
					<compilerVersion>1.7</compilerVersion>
				</configuration>
			</plugin>
		</plugins>
	</build>

	<repositories>

		<!-- Repository where we can found the storm dependencies -->
		<repository>
			<id>clojars.org</id>
			<url>http://clojars.org/repo</url>
		</repository>

	</repositories>

	<dependencies>
		<dependency>
			<groupId>com.alibaba.jstorm</groupId>
			<artifactId>jstorm-core</artifactId>
			<version>2.1.1</version>
			<scope>provided</scope>
			<exclusions>
				<exclusion>
					<artifactId>slf4j-nop</artifactId>
					<groupId>org.slf4j</groupId>
				</exclusion>
				<exclusion>
					<artifactId>slf4j-jdk14</artifactId>
					<groupId>org.slf4j</groupId>
				</exclusion>
			</exclusions>

		</dependency>
		<dependency>
			<groupId>com.twitter</groupId>
			<artifactId>chill-java</artifactId>
			<version>0.3.5</version>
		</dependency>

	</dependencies>

</project>

import spouts.WordReader;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import bolts.WordCounter;
import bolts.WordNormalizer;

public class TopologyMain {
	public static void main(String[] args) throws InterruptedException {

		// Topology definition(定义Topology)
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("word-reader", new WordReader());
		builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader");
		builder.setBolt("word-counter", new WordCounter(), 1).fieldsGrouping("word-normalizer", new Fields("word"));

		// Configuration
		Config conf = new Config();
		conf.put("wordsFile", "src/main/resources/words.txt");//配置参数自定义
		conf.setDebug(false);
		conf.setNumWorkers(1);//分配多少个Worker进程去执行这个Topology
		// Topology run
		conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);//SPOUT最大Tuple缓冲区,大于就挂起,等待bolt消费小于这个值再运行

		//本地模式
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology());
		Thread.sleep(20000);
		cluster.shutdown();

		//服务器模式
		//StormSubmitter.submitTopology("Getting-Started-Toplogie", conf, builder.createTopology());
	}
}

package spouts;

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;

public class WordReader extends BaseRichSpout {
	private SpoutOutputCollector collector; // 保存open时存进来的SpoutOutputCollector，要用它向Bolt发送Tuple
	private boolean completed = false;

	private String[] readerString = new String[] { "test10", "test20", "test30" };
	private String[] readerCount = new String[] { "10", "20", "30" };

	public void ack(Object msgId) {
		System.out.println("WordReader OK:" + msgId);
	}

	public void close() {
		System.out.println("WordReader close");
	}

	public void fail(Object msgId) {
		System.out.println("WordReader FAIL:" + msgId);
	}

	public void nextTuple() {
		String string;
		String count;
		int messageId;

	if (completed) {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			return;
		}

		for (int i = 0; i < 3; i++) {
			string = readerString[i];
			count = readerCount[i];
			messageId = i;
			this.collector.emit(new Values(string, count), messageId);
		}

		completed = true;
	}

	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		System.out.println("WordReader open:");
		this.collector = collector;
	}

	// 定义map中的key,方便new Values时不用再写key了，位置一一对应
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		System.out.println("WordReader declareOutputFields:");
		declarer.declare(new Fields("string", "count"));
	}
}

package bolts;

import java.util.Map;

import backtype.storm.task.TopologyContext;
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;
import clojure.lang.IRecord;

public class WordNormalizer extends BaseBasicBolt {

	public void cleanup() {
		System.out.println("WordNormalizer cleanup");
	}

	public void prepare(Map stormConf, TopologyContext context) {
		System.out.println("WordNormalizer prepare");
		System.out.println("WordNormalizer prepare wordsFile == " + stormConf.get("wordsFile").toString());
		System.out.println("WordNormalizer prepare2");
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		String string = input.getString(0);
		String count = input.getString(1);// 因为WordReader只有两个域
		System.out.println("WordNormalizer execute string == " + string);
		System.out.println("WordNormalizer execute count == " + count);
		collector.emit(new Values(count, string));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		System.out.println("WordNormalizer declareOutputFields");
		declarer.declare(new Fields("count", "string"));
	}
}

package bolts;

import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class WordCounter extends BaseBasicBolt {

	Integer id;
	String name;

	@Override
	public void cleanup() {
		System.out.println("WordCounter cleanup");
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		System.out.println("WordCounter prepare");
		System.out.println("WordCounter prepare wordsFile == " + stormConf.get("wordsFile").toString());
		this.name = context.getThisComponentId();
		this.id = context.getThisTaskId();
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// 这里是最后一个Bolt不用定义了
		System.out.println("WordCounter declareOutputFields");
	}

	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		String string = input.getString(0);
		String count = input.getString(1);// 因为WordReader只有两个域
		System.out.println("WordCounter execute string == " + string);
		System.out.println("WordCounter execute count == " + count);
	}
}

参考原文：http://www.open-open.com/lib/view/open1430095563146.html
参考原文：http://ifeve.com/getting-started-with-stom-index/
参考原文：https://github.com/runfriends/GettingStartedWithStorm-cn/blob/master/chapter2/Hello%20World%20Storm.md#%E5%88%9B%E5%BB%BA%E6%88%91%E4%BB%AC%E7%9A%84%E7%AC%AC%E4%B8%80%E4%B8%AA%E6%8B%93%E6%89%91
参考原文：http://blog.itpub.net/29754888/viewspace-1260026/
参考原文：https://book.douban.com/reading/33008852/
参考原文：http://shiyanjun.cn/archives/977.html


相关jar包: