Storm (实时分布式大数据处理系统) 简介

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
云原生网关 MSE Higress,422元/月
简介: 相比Hadoop的批处理,Storm的特点就是实时性。 组件 Storm集群主要由一个主节点和一群工作节点(worker node)组成,通过 Zookeeper进行协调。主节点 主节点通常运行一个后台程序 —— Nimbus,用于响应分布在集群中的节点,分配任务和监测故障。这个很类似于Hadoop中的Job Tracker。工作节点 工作节点叫worker,一般就是集群中的一个节

相比Hadoop的批处理,Storm的特点就是实时性。

组件

Storm集群主要由一个主节点和一群工作节点(worker node)组成,通过 Zookeeper进行协调。

主节点
主节点通常运行一个后台程序 —— Nimbus,用于响应分布在集群中的节点,分配任务和监测故障。这个很类似于Hadoop中的Job Tracker。

工作节点
工作节点叫worker,一般就是集群中的一个节点,也就是一个计算机。它同样会运行一个后台程序 ——Supervisor,用于收听工作指派并基于要求运行工作进程。每个工作节点都是topology中一个子集的实现。而Nimbus和Supervisor之间的协调则通过Zookeeper系统或者集群。

topology
[tə'pɒlədʒɪ]
topology是一组由Spouts(数据源)和Bolts(数据操作)通过Stream Groupings进行连接的图。也就是我们的整个应用程序。

Zookeeper
Zookeeper是完成Supervisor和Nimbus之间协调的服务。而应用程序实现实时的逻辑则被封装进Storm中的“topology”。

Spout
[spaʊt]  n. 喷水口
简而言之,Spout从来源处读取数据并放入topology。Spout分成可靠和不可靠两种;当Storm接收失败时,可靠的Spout会对tuple(元组,数据项组成的列表)进行重发;而不可靠的Spout不会考虑接收成功与否只发射一次。而Spout中最主要的方法就是nextTuple(),该方法会发射一个新的tuple到topology,如果没有新tuple发射则会简单的返回。

Bolt
[bəʊlt] n.门闩

Topology中所有的处理都由Bolt完成。Bolt可以完成任何事,比如:连接的过滤、聚合、访问文件/数据库、等等。Bolt从Spout中接收数据并进行处理,如果遇到复杂流的处理也可能将tuple发送给另一个Bolt进行处理。而Bolt中最重要的方法是execute(),以新的tuple作为参数接收。不管是Spout还是Bolt,如果将tuple发射成多个流,这些流都可以通过declareStream()来声明。

并行参数

worker,进程数。

Executor,线程总数。

Task,具体的spout和bolt的实例个数。一个Executor可以负责1个或多个task。一般地,task数等于executor数。

tuple分组策略

分布式处理的主要思想就是把大的任务划分成多个子任务,它们在不同的机器或线程中并行处理,最后汇合。以统计班上同学的平均成绩为例,说明问题。
微笑bolt如何设计?
因为分布式的原因,同一bolt类的多个线程之间不能有数据共享。分十个机器,编号为i的机器统计学号末尾为i的同学数和他们的总成绩,最后汇总。
微笑如何控制每一个bolt实例接收哪些学号的数据呢?
结合上文,编号为i的bolt实例只接受学号末尾为i的数据。可以按照tuple的字段分组。

tuple传递是通过序列化,套接字传输,反序列化实现的。
所谓的grouping策略就是在Spout与Bolt、Bolt与Bolt之间传递Tuple的方式。总共有七种方式:
1)shuffleGrouping(随机分组)
2)fieldsGrouping(按照字段分组,在这里即是同一个单词只能发送给一个Bolt)

//declare与emit是一一对应的。
declarer.declareStream("http", new Fields("event", "key"));
collector.emit("http", new Values(new MyClass(), key));
//设置bolt的时候指定按照哪个字段散列
builder.setBolt("httpCount", new AnalyseBolt()).fieldsGrouping("XX", "http", new Fields("key"));
//接收到tuple时按指定字段取值
public void process(Tuple tuple, BasicOutputCollector collector) {
	tuple.getValueByField("event") instanceof MyClass
}	

3)allGrouping(广播发送,即每一个Tuple,每一个Bolt都会收到)
4)globalGrouping(全局分组,将Tuple分配到task id值最低的task里面)
5)noneGrouping(随机分派)
6)directGrouping(直接分组,指定Tuple与Bolt的对应发送关系)
7)Local or shuffle Grouping

8)customGrouping (自定义的Grouping)

消息可靠性交付

要理解这个问题,需要看一下tuple在离开spout之后的生命周期。作为参考,下面是spout实现的接口

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

首先,Storm会通过Spout的 nextTuple()方法从Spout申请一个tuple。在open方法中,Spout使用此方法提供的SpoutOutputCollector去发射一个tuple到输出streams中去。当发射一个tuple时,Spout会提供一个“message id”,用来后面区分不同的tuple。例如, KestrelSpout从kestrel队列中读取消息,然后在发射时会将Kestrel为消息提供的id作为“message id”。发射一条消息到SpoutOutputCollector,如下所示:
_collector. emit(newValues("field1", "field2", 3), msgId);
然后,这个tuple会发送到消费bolts,同时Storm会跟踪已被创建的消息树状图。如果Storm检测到一个tuple已被“fully processed”, Storm将会原始的Spout task(即发射这个tuple的Spout)上调用ack()方法,参数msgId就是这个Spout提供给Storm的“message id”。类似的,如果这个tuple超时了, Storm会在原始的Spout task上调用fail()方法。注意, 一个tuple只能被创建它的Spouttask进行acked或者failed。因此,即使一个Spout在集群上正在执行很多tasks,一个tuple也只能被创建它的task进行acked或failed,而其他的task则不行。

storm UI

storm自带的一个仪表盘,见图1.


图1 storm自带的一个仪表盘

emitted,发射的tuple数。
transferred,若一个tuple被其他bolt读取,则transferred+1。
acked,tuple被完整处理。
failed,在处理过程中出现错误或超时的tuple数。超时参数可以设置。


相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
相关文章
|
13天前
|
存储 缓存 分布式计算
大数据-83 Spark 集群 RDD编程简介 RDD特点 Spark编程模型介绍
大数据-83 Spark 集群 RDD编程简介 RDD特点 Spark编程模型介绍
29 4
|
14天前
|
缓存 NoSQL Java
大数据-50 Redis 分布式锁 乐观锁 Watch SETNX Lua Redisson分布式锁 Java实现分布式锁
大数据-50 Redis 分布式锁 乐观锁 Watch SETNX Lua Redisson分布式锁 Java实现分布式锁
37 3
大数据-50 Redis 分布式锁 乐观锁 Watch SETNX Lua Redisson分布式锁 Java实现分布式锁
|
8天前
|
消息中间件 中间件 数据库
NServiceBus:打造企业级服务总线的利器——深度解析这一面向消息中间件如何革新分布式应用开发与提升系统可靠性
【10月更文挑战第9天】NServiceBus 是一个面向消息的中间件,专为构建分布式应用程序设计,特别适用于企业级服务总线(ESB)。它通过消息队列实现服务间的解耦,提高系统的可扩展性和容错性。在 .NET 生态中,NServiceBus 提供了强大的功能,支持多种传输方式如 RabbitMQ 和 Azure Service Bus。通过异步消息传递模式,各组件可以独立运作,即使某部分出现故障也不会影响整体系统。 示例代码展示了如何使用 NServiceBus 发送和接收消息,简化了系统的设计和维护。
22 3
|
13天前
|
SQL 关系型数据库 分布式数据库
Citus 简介,将 Postgres 转换为分布式数据库
【10月更文挑战第4天】Citus 简介,将 Postgres 转换为分布式数据库
57 4
|
14天前
|
存储 缓存 NoSQL
大数据-38 Redis 高并发下的分布式缓存 Redis简介 缓存场景 读写模式 旁路模式 穿透模式 缓存模式 基本概念等
大数据-38 Redis 高并发下的分布式缓存 Redis简介 缓存场景 读写模式 旁路模式 穿透模式 缓存模式 基本概念等
37 4
|
14天前
|
缓存 NoSQL Ubuntu
大数据-39 Redis 高并发分布式缓存 Ubuntu源码编译安装 云服务器 启动并测试 redis-server redis-cli
大数据-39 Redis 高并发分布式缓存 Ubuntu源码编译安装 云服务器 启动并测试 redis-server redis-cli
41 3
|
14天前
|
存储 开发框架 .NET
C#语言如何搭建分布式文件存储系统
C#语言如何搭建分布式文件存储系统
47 2
|
4天前
|
消息中间件 存储 监控
消息队列系统中的确认机制在分布式系统中如何实现?
消息队列系统中的确认机制在分布式系统中如何实现?
|
14天前
|
存储 SQL 消息中间件
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
35 0
|
6天前
|
NoSQL Java Redis
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
Redis分布式锁在高并发场景下是重要的技术手段,但其实现过程中常遇到五大深坑:**原子性问题**、**连接耗尽问题**、**锁过期问题**、**锁失效问题**以及**锁分段问题**。这些问题不仅影响系统的稳定性和性能,还可能导致数据不一致。尼恩在实际项目中总结了这些坑,并提供了详细的解决方案,包括使用Lua脚本保证原子性、设置合理的锁过期时间和使用看门狗机制、以及通过锁分段提升性能。这些经验和技巧对面试和实际开发都有很大帮助,值得深入学习和实践。
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?

热门文章

最新文章