Seal^_^头像
关注
Storm架构深度拆解：Nimbus、Supervisor、Worker、Executor与Task模型封面图

Storm架构深度拆解：Nimbus、Supervisor、Worker、Executor与Task模型

Storm架构深度拆解：Nimbus、Supervisor、Worker、Executor与Task模型


1. Storm架构概述与核心组件


Apache Storm是一个开源的分布式实时计算系统，专为处理大规模实时数据流而设计。与传统的批处理系统不同，Storm强调低延迟处理能力，能够以毫秒级速度处理源源不断的数据流。在深入解析Storm的内部架构之前，我们首先需要了解其核心组件及其相互关系。


Storm架构主要由以下几个核心组件构成：


  1. Nimbus：集群的"大脑"，负责任务调度、资源分配和监控。它类似于Hadoop中的JobTracker，但不负责实际的数据处理。


  1. Supervisor：集群的"执行者"，在每个物理节点上运行，负责管理该节点上的Worker进程。每个Supervisor节点可以根据配置运行多个Worker。


  1. Worker：在Supervisor上运行的进程，负责执行具体的拓扑(Topology)部分。每个Worker可以包含多个Executor。


  1. Executor：在Worker内部运行的工作线程，负责执行一个或多个Task。Executor的数量可以在拓扑定义时指定。


  1. Task：实际执行数据处理的逻辑单元，对应Bolt或Spout中的特定逻辑。一个Executor可以运行多个Task，这些Task运行在同一个线程中。


这些组件共同构成了Storm的分布式计算框架，通过合理配置它们之间的关系，可以实现高效、稳定的实时数据处理。


Storm架构整体图展示Storm架构中Nimbus、Supervisor、Worker、Executor和Task之间的关系NimbusSupervisor 1Supervisor 2Supervisor 3Worker 1Worker 2Worker 1Worker 2Worker 3Executor 1Executor 2Executor 1Executor 2Executor 1Task 1Task 2Task 3Task 1Task 2Task 3Task 4Task 1Task 2


上图展示了Storm的整体架构关系：Nimbus作为中心节点，向各个Supervisor节点分配任务；每个Supervisor管理多个Worker进程；每个Worker内部包含多个Executor线程；每个Executor可以处理多个Task。这种分层结构确保了系统的高可用性和可扩展性。


2. Nimbus与Supervisor的协调机制


Nimbus和Supervisor是Storm集群的核心协调组件，它们之间的有效协作是整个系统稳定运行的关键。理解它们的交互机制对于优化Storm集群性能和排查故障至关重要。


Nimbus的主要职责包括：

  • 接收用户提交的拓扑(Topology)定义
  • 将拓扑分解为任务，并根据集群资源情况进行分配
  • 监控集群中各个节点的状态和任务执行情况
  • 在节点故障时重新分配任务


Supervisor的主要职责包括：

  • 监听来自Nimbus的任务分配指令
  • 管理本机上的Worker进程生命周期
  • 向Nimbus报告本地资源使用情况和Worker状态
  • 在Worker进程异常时尝试重启


Nimbus和Supervisor之间的通信采用ZooKeeper进行协调，确保在Nimbus节点故障时，新的N节点可以无缝接管集群管理。这种设计使得Storm具有高可用性，即使部分节点发生故障，系统也能继续提供服务。


Nimbus与Supervisor交互流程展示Nimbus和Supervisor之间的交互流程，包括任务分发、监控和故障处理用户NimbusSupervisorWorker进程Executor线程ZooKeeper提交拓扑任务分配状态监控故障处理资源管理


上图展示了Nimbus与Supervisor之间的交互流程：用户提交拓扑到Nimbus，Nimbus通过ZooKeeper将任务分配给Supervisor，Supervisor管理Worker进程和Executor线程，同时进行状态监控和资源管理，并在需要时处理故障。这种分布式协调机制确保了系统的高可用性和可靠性。


3. Worker、Executor与Task的三级模型


理解Worker、Executor和Task的三级模型对于优化Storm拓扑的性能至关重要。这三层结构共同构成了Storm并行计算的基础，每一层都有特定的职责和配置方式。


Worker层


Worker是由Supervisor创建并管理的进程，它是运行拓扑的基本单元。每个Worker在单独的JVM中运行，隔离了不同Worker之间的资源使用和故障影响。Worker的数量是在提交拓扑时通过worker.num参数指定的。


Worker的主要职责包括：

  • 接收来自Supervisor的资源分配
  • 加载和管理指定的Executor
  • 提供JVM环境和资源隔离
  • 处理Executor之间的通信


每个Worker只能处理拓扑的一部分，一个完整的拓扑通常分布在多个Worker中，这种设计增强了系统的可扩展性和容错性。


Executor层


Executor是在Worker内部运行的工作线程，负责执行一个或多个Task。Executor的数量可以在拓扑定义时通过parallelism_hint参数指定，但实际创建的Executor数量可能会根据集群资源情况进行调整。


Executor的主要职责包括：

  • 执行Task的具体处理逻辑
  • 管理Task之间的消息传递
  • 处理来自上游Tuple的流
  • 实现负载均衡机制


值得注意的是，每个Executor可以在多个Task之间共享资源，但一次只能处理一个Tuple，这是为了保证处理的顺序性和一致性。


Task层


Task是执行数据处理的最小逻辑单元，对应Bolt或Spout中的特定实例。Task的数量由拓扑中所有组件的parallelism_hint乘积决定，实际执行的Task数量可能会根据集群情况进行调整。


Task的主要职责包括：

  • 执行具体的数据处理逻辑
  • 维护状态信息
  • 发送Tuple到下游组件
  • 处理ACK和FAIL消息


这种三级模型提供了极大的灵活性，允许开发者根据业务需求并行处理数据流，同时保持资源使用的效率。


Worker、Executor与Task的三级结构展示Worker、Executor和Task之间的关系，以及它们如何并行处理任务Worker进程Executor 1Executor 2Executor 3Task ATask BTask CTask DTask ETask FTask G状态1状态2状态3状态4状态5状态6状态7状态8数据流处理：从Spout到Bolt的Tuple传递与状态更新


上图清晰地展示了Worker、Executor和Task的三级结构：Worker作为JVM进程容器，包含多个Executor线程；每个Executor可以处理多个Task；每个Task维护自己的状态信息，并在处理过程中更新这些状态。这种分层结构提供了良好的资源隔离和并行处理能力。


4. Storm任务调度与并行处理


Storm的任务调度机制是其高效处理实时数据流的核心。了解Nimbus如何调度任务、Supervisor如何分配资源，以及Worker、Executor和Task如何协同工作，对于优化拓扑性能至关重要。


任务调度过程


Nimbus在收到拓扑提交后，会执行以下调度步骤：


  1. 解析拓扑定义：解析拓扑中所有组件的并行度(Parallelism)和配置参数。


  1. 计算资源需求：根据组件的并行度计算所需的Task数量，以及所需的Executor和Worker数量。


  1. 评估集群资源：检查集群中各个Supervisor节点的资源情况(CPU、内存等)。


  1. 任务分配：将Task分配到合适的Supervisor节点，并确保负载均衡。


  1. 部署执行：向Supervisor发送指令，启动相应的Worker和Executor。


这一过程通常在几秒内完成，因此Storm能够快速适应集群负载的变化。


并行处理机制


Storm的并行处理基于以下机制：


  1. 并行度控制：通过设置组件的parallelism_hint参数，控制Task的数量。


  1. 分区策略：对于有状态的操作，可以通过分区策略确保相关数据被发送到同一个Task。


  1. 资源分配：每个Worker和Executor可以配置使用多少CPU和内存资源。


  1. 负载均衡：Storm内置了多种负载均衡策略，可以根据数据特征选择合适的策略。


合理的并行配置可以显著提升处理性能，过高或过低的并行度都会影响效率。


容错机制


Storm通过以下机制确保任务的容错性：


  1. 消息确认机制：每个Tuple都需要被处理完成后才能被确认，未确认的Tuple会被重新发送。


  1. 超时重试：对于长时间未响应的Tuple，Storm会触发超时机制并重新发送。


  1. 故障转移：当Task或Executor失败时，Nimbus会在其他节点上重新启动它。


  1. 检查点机制：支持定期保存状态到持久存储，实现故障恢复。


Storm任务调度决策树展示Storm如何根据资源情况调度任务，包括决策点和相应的处理流程提交拓扑是否解析拓扑定义计算资源需求资源不足资源充足完成等待资源评估集群资源开始执行超时资源紧张资源充裕监控拒绝提交扩容集群任务分配部署执行重新提交任务监控运行中


上图展示了Storm任务调度的决策树：从提交拓扑开始，系统会解析拓扑定义并计算资源需求；如果资源不足，可以选择等待资源或扩容集群；如果资源充足，则评估集群资源并决定任务分配策略；最后部署执行并进行监控。这一过程确保了任务的高效调度和集群的稳定运行。


5. 实战示例与最佳实践


理论讲解后，我们通过一个简单的WordCount示例来演示如何构建一个Storm拓扑，并分享一些配置和优化方面的最佳实践。


示例：WordCount拓扑实现


以下是一个基本的WordCount拓扑实现：


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;

public class WordCountTopology {
    public static class WordSpout extends BaseRichSpout {
        // 实现Spout逻辑，读取文本数据并发射单词
        // ...
    }
    
    public static class WordSplitter extends BaseRichBolt {
        // 实现Splitter逻辑，将句子拆分为单词
        // ...
    }
    
    public static class WordCounter extends BaseRichBolt {
        // 实现Counter逻辑，统计单词频率
        // ...
    }
    
    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        
        // 设置Spout并行度为2
        builder.setSpout("word-spout", new WordSpout(), 2);
        
        // 设置Splitter并行度为4
        builder.setBolt("word-splitter", new WordSplitter(), 4)
               .fieldsGrouping("word-spout", new Fields("word"));
        
        // 设置Counter并行度为6
        builder.setBolt("word-counter", new WordCounter(), 6)
               .fieldsGrouping("word-splitter", new Fields("word"));
        
        // 创建配置
        Config conf = new Config();
        conf.setDebug(true);
        conf.setNumWorkers(3); // 设置Worker数量为3
        
        // 本地模式运行
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-count", conf, builder.createTopology());
        
        Utils.sleep(10000);
        cluster.shutdown();
    }
}


在这个示例中，我们定义了三个组件：WordSpout负责读取文本数据并发射单词；WordSplitter负责将句子拆分为单词；WordCounter负责统计单词频率。通过设置不同的并行度，我们可以控制任务的并行处理能力。


最佳实践


  1. 合理的并行度设置：
  • 根据数据量和集群资源设置合适的并行度
  • Spout的并行度通常不需要太高，因为数据源往往是瓶颈
  • Bolt的并行度可以根据处理能力适当增加


  1. 资源分配优化：
  • 为Worker分配足够的内存，避免JVM溢出
  • 根据任务复杂度调整CPU资源分配
  • 监控资源使用情况，及时调整配置


  1. 容错与可靠性配置：
  • 配置合适的消息超时和重试次数
  • 对于有状态的操作，考虑使用可靠的持久化存储
  • 实现自定义的故障处理逻辑


  1. 性能调优：
  • 减少不必要的序列化和反序列化
  • 使用批量处理减少网络开销
  • 优化Tuple结构，减少数据传输量


Storm配置优化对比展示不同配置下Storm的性能表现，包括吞吐量、延迟等指标配置方案A配置方案B配置方案C推荐性能指标:低并行度中等并行度高并行度最优并行度吞吐量: 10K/s吞吐量: 25K/s吞吐量: 20K/s吞吐量: 30K/s延迟: 50ms延迟: 30ms延迟: 60ms延迟: 25ms资源利用率: 40%资源利用率: 70%资源利用率: 85%资源利用率: 80%配置方案C资源利用率过高，可能导致系统不稳定；最优并行度平衡了吞吐量、延迟和资源利用率


上图比较了不同配置方案下的性能表现：配置方案A(低并行度)虽然资源利用率低，但吞吐量也较低；配置方案B(中等并行度)在各方面表现均衡；配置方案C(高并行度)资源利用率过高，可能导致系统不稳定；最优并行度在各方面都有最佳表现。实际应用中，需要根据业务需求找到最合适的平衡点。


注意事项


  1. 内存管理：
  • 确保为Worker分配足够的堆内存，避免频繁GC
  • 避免在Spout或Bolt中缓存大量数据，使用外部存储代替
  • 考虑使用Off-Heap内存减少GC压力


  1. 消息处理：
  • 避免在Tuple中传输大对象，减少序列化开销
  • 合理设置批处理大小，平衡延迟和吞吐量
  • 使用合理的分组策略，减少数据倾斜


  1. 故障处理：
  • 实现可靠的输出策略，确保数据不丢失
  • 考虑使用Storm的Trident API简化状态管理
  • 实现自定义的监控和报警机制


通过遵循以上实践和注意事项，可以构建出高效、稳定的实时数据处理系统。


最小示例与运行注意事项


下面是一个可直接运行的Storm最小示例，展示了一个简单的单词计数拓扑：


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

import java.util.Arrays;
import java.util.List;

public class SimpleWordCount {
    public static class RandomSentenceSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;
        private int sentenceIndex = 0;
        private List<String> sentences = Arrays.asList(
            "the cow jumped over the moon",
            "an apple a day keeps the doctor away",
            "four score and seven years ago"
        );

        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void nextTuple() {
            Utils.sleep(100);
            String sentence = sentences.get(sentenceIndex % sentences.size());
            collector.emit(new Values(sentence));
            sentenceIndex++;
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("sentence"));
        }
    }

    public static class SplitSentenceBolt extends BaseRichBolt {
        private OutputCollector collector;

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

        @Override
        public void execute(Tuple tuple) {
            String sentence = tuple.getStringByField("sentence");
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }

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

    public static class WordCountBolt extends BaseRichBolt {
        private Map<String, Integer> counts = new HashMap<>();
        private OutputCollector collector;

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

        @Override
        public void execute(Tuple tuple) {
            String word = tuple.getStringByField("word");
            Integer count = counts.get(word);
            if (count == null) {
                count = 0;
            }
            count++;
            counts.put(word, count);
            collector.emit(new Values(word, count));
        }

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

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();
        
        // 设置Spout并行度为1，仅用于演示
        builder.setSpout("spout", new RandomSentenceSpout(), 1);
        
        // 设置Splitter并行度为2
        builder.setBolt("split", new SplitSentenceBolt(), 2)
               .shuffleGrouping("spout");
        
        // 设置Counter并行度为2
        builder.setBolt("count", new WordCountBolt(), 2)
               .fieldsGrouping("split", new Fields("word"));
        
        // 配置
        Config conf = new Config();
        conf.setDebug(true);
        conf.setNumWorkers(2);
        
        // 本地模式运行
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("word-count", conf, builder.createTopology());
        
        // 运行10秒后关闭
        Utils.sleep(10000);
        cluster.shutdown();
    }
}


运行注意事项：


  1. 依赖：确保项目中包含Storm核心依赖：

```xml

<dependency>

<groupId>org.apache.storm</groupId>

<artifactId>storm-core</artifactId>

<version>1.2.2</version>

</dependency>

```


  1. 本地 vs 集群模式：示例使用LocalCluster运行，生产环境应配置为集群模式，包括Nimbus、Supervisor等组件的正确部署。


  1. 资源配置：根据集群规模调整worker数量和并行度，避免资源不足或浪费。


  1. 监控：添加适当的日志记录，便于排查问题和监控系统运行状态。


  1. 容错：对于生产环境，需要实现可靠的输出策略和适当的容错机制。


通过这个示例，您可以快速理解Storm的基本工作流程，并根据实际需求进行扩展和优化。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/qq_41840843/article/details/166690483

文章来源转载

评论

赞0

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--