Storm大数据位置
Storm
流式处理
Twitter 的新流式大数据分析解决方案
S4
Yahoo!
流式处理
来自 Yahoo! 的分布式流计算平台
Hadoop
Apache
批处理
MapReduce 范式的第一个开源实现
Spark
UC Berkeley AMPLab
批处理/流处理
支持内存中数据集和恢复能力的最新分析平台
Disco
Nokia
批处理
Nokia 的分布式 MapReduce 框架
HPCC
LexisNexis
批处理
HPC 大数据集群
二、Storm概念及组件
6. Spout：在一个topology中产生源数据流的组件。通常情况下spout会从外部数据源中读取数据，然后转换为topology内部的源数据。Spout是一个主动的角色，其接口中有个nextTuple()函数，storm框架会不停地调用 此函数，用户只要在其中生成源数据即可。
7. Bolt： 在一个topology中接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、写数据库等任何操作。Bolt是一个被动的角色，其接口中有个execute(Tuple input)函数,在接受到消息后会调用此函数，用户可以在其中执行自己想要的操作。
8. Tuple： Storm Spout、Bolt组件消息传递的基本单元（数据模型），Tuple是包含名称的列表，Storm支持所有原生类型，字节数组为Tuple字段 传递，如果要传递自定义对象，需要实现接口serializer。
三、创建逻缉组件
下图二所示，创建Storm组件Spout、Bolt及Storm拓扑构建并结合拓扑描述并发情况（Worker、Executor、Task关系）
SPOUT：自定义类实现IRichSpout接口或继承BaseRichSpout即创建Spout组件。Spout从外部数据源读取数据（队列、DRPC等）为Storm拓扑提供数据，Storm可以实现可靠或不可靠，可靠性表现在Storm可以对Storm toplogy处理失败的tuple进行重发，反之不处理。
Spout可以发送多个流，通过在调用SpoutOutputCollector类的emit方法的同时使用declareStream方法申 明并指定多个流发送。
Spout中主要方法nextTuple，能够发送新流到toplopy及无数据流发送时直接返回。最重要的是不要阻塞 nextTuple方法，因为Spout在同一个线程执行。Spout主要将读取到的数据组织成Tuple发送到Bolt组件处 理。
public class SimpleTopolog {
public static void main(String[] args) throws Exception {
// 解决ZOOKEEPER客户端连接服务端问题(IPV6)
System.setProperty("java.net.preferIPv4Stack", "true");
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("green-spout", new GreenSpout(), 2) // 2 executor(线程)
.setNumTasks(4); // 4 task 对应2个executor
builder.setBolt("yellow-bolt", new YellowBolt(),6) // 6 executor(线程)
.shuffleGrouping("green-spout"); //tuple随机分发到bolt处理
builder.setBolt("blue-bolt", new BlueBolt(), 2) // 2 executor(线程)
.shuffleGrouping("yellow-bolt");
Config conf = new Config();
/*设置工作进程数*/
conf.setNumWorkers(2);
conf.setDebug(true);
/*本地运行模式*/
StormRunner.runTopologyLocally(builder.createTopology(), "simpleTopology", conf, 0);
}
————————————————
package com.sunshine.tools;
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.generated.StormTopology;
public final class StormRunner {
private static final int MILLIS_IN_SEC = 1000;
private StormRunner() {
}
public static void runTopologyLocally(StormTopology topology, String topologyName, Config conf, int runtimeInSeconds)
throws InterruptedException {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(topologyName, conf, topology);
Thread.sleep((long) runtimeInSeconds * MILLIS_IN_SEC);
cluster.killTopology(topologyName);
cluster.shutdown();
}
public static void runTopologyRemotely(StormTopology topology, String topologyName, Config conf)
throws AlreadyAliveException, InvalidTopologyException {
StormSubmitter.submitTopology(topologyName, conf, topology);
}
}
推荐阅读：
https://blog.csdn.net/yangyangmyself/article/details/52346859
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/69949806/viewspace-2914509/ ，如需转载，请注明出处，否则将追究法律责任。
