Apache Storm 动态扩缩容：Rebalance、在线调整并行度与拓扑滚动升级实战- 惊觉

Apache Storm 动态扩缩容：Rebalance、在线调整并行度与拓扑滚动升级实战

1. Storm 动态扩缩容概述

Apache Storm 是一个开源的分布式实时计算系统，广泛应用于流数据处理场景。在生产环境中，随着业务负载的变化，需要对 Storm 集群进行动态扩缩容以优化资源利用率并保持系统性能。Storm 提供了几种机制来实现动态扩缩容，主要包括 Rebalance 命令、在线调整并行度以及拓扑滚动升级。

上图展示了 Storm 集群的典型架构以及实现动态扩缩容的核心组件。Nimbus 主节点负责资源分配和任务调度，Supervisor 工作节点托管实际执行计算的 Worker 进程，而每个 Worker 包含多个 Executor 线程来处理具体的计算任务。ZooKeeper 作为协调服务存储集群状态和拓扑信息，而底部的三种扩缩容机制则提供了不同的动态调整能力。

Storm 动态扩缩容的主要优势在于：

无需停止拓扑即可进行资源调整
支持平滑过渡，避免服务中断
可根据实际负载实时优化资源配置
提供细粒度的并行度控制能力

2. Rebalance 机制详解

Rebalance 是 Storm 提供的一种核心动态扩缩容机制，允许在不停止拓扑的情况下重新分配计算资源。通过 Rebalance，我们可以动态调整拓扑中各组件的并行度，以应对负载变化。

Rebalance 工作原理

Rebalance 命令通过以下步骤实现资源重新分配：

向 Nimbus 提交 Rebalance 请求
Nimbus 通知所有相关 Supervisor 准备迁移
Supervisor 重新分配 Executor 进程
重新分配数据流和分组策略

上图展示了 Rebalance 机制的执行决策树。当提交 Rebalance 请求后，系统会判断是首次执行还是二次执行。如果是首次执行，系统会询问是否需要重置 Executor 数量；如果是二次执行，则会检查是否有未处理消息。根据不同的决策路径，系统会采取相应操作，最终实现资源的重新分配。

Rebalance 命令使用示例

# 基本语法
storm rebalance <topology-name> [option1] [option2] ...

# 示例：将并行度调整为 10，等待 30 秒
storm rebalance my-topology -n 10 -w 30

# 示例：只重新分配 bolt 的并行度
storm rebalance my-topology -e bolt:5

# 示例：强制重新分配（丢弃未处理消息）
storm rebalance my-topology --force

Rebalance 关键参数解释

参数	描述	默认值	推荐场景
`-n`, `--num-executors`	设置 Executor 总数	保持当前值	统一增加/减少资源
`-e`, `--executors`	按组件设置 Executor 数量	保持当前值	精细化调整特定组件
`-w`, `--wait`	等待时间(秒)	0	需要优雅停止的情况
`--force`	强制重新分配	false	需要立即生效的场景

3. 在线调整并行度实践

除了 Rebalance 机制外，Storm 还提供了更细粒度的并行度调整能力，允许单独调整拓扑中各个组件（Spout/Bolt）的并行度。

并行度调整的工作原理

并行度调整通过修改拓扑配置中的 parallelism.hint 参数实现。当调整并行度后，Nimbus 会：

计算需要的 Executor 数量
分配新的资源
启动新 Executor 并迁移状态

上图展示了并行度调整的完整流程。首先，系统会解析新的并行度配置，然后计算所需的资源需求，分配新资源并启动新的 Executor 进程，最后迁移数据流并使新并行度生效。

并行度调整实现方法

方法一：通过 Storm UI 调整

登录 Storm Web UI (http://<nimbus-host>:8080)
选择目标拓扑
点击 "Rebalance" 按钮
在弹出的对话框中调整并行度参数
提交调整请求

方法二：通过代码动态调整

// 创建拓扑配置
TopologyBuilder builder = new TopologyBuilder();
// 设置初始并行度
builder.setSpout("my-spout", new MySpout(), 2);
builder.setBolt("my-bolt1", new MyBolt1(), 3)
       .shuffleGrouping("my-spout");
builder.setBolt("my-bolt2", new MyBolt2(), 2)
       .shuffleGrouping("my-bolt1");

// 创建配置
Config conf = new Config();
conf.setNumWorkers(4);
conf.setMaxTaskParallelism(12);

// 提交拓扑
StormSubmitter.submitTopology("my-topology", conf, builder.createTopology());

// 后续动态调整并行度的代码
// 获取拓扑配置
TopologyConfig topologyConfig = new TopologyConfig();
topologyConfig.setComponentParallelism("my-spout", 4); // 调整 spout 并行度为 4
topologyConfig.setComponentParallelism("my-bolt1", 5); // 调整 bolt1 并行度为 5
topologyConfig.setComponentParallelism("my-bolt2", 3); // 调整 bolt2 并行度为 3

// 提交调整请求
StormSubmitter.rebalance("my-topology", topologyConfig, 30); // 等待 30 秒完成调整

方法三：通过 Storm CLI 命令调整

# 设置拓扑配置文件
cat > topology-config.yaml <<EOF
config:
  spout:
    parallelism: 4
  bolt1:
    parallelism: 5
  bolt2:
    parallelism: 3
  workers: 6
EOF

# 应用配置调整
storm rebalance my-topology -f topology-config.yaml

并行度调整最佳实践

逐步调整：避免一次性大幅度调整并行度
监控资源：调整前后密切关注 CPU、内存使用情况
数据倾斜：特别关注数据倾斜问题，避免部分负载过高
状态管理：有状态的组件调整并行度需要谨慎处理状态迁移
优雅过渡：设置适当的等待时间，确保数据处理完整

4. 拓扑滚动升级策略

滚动升级是在不中断服务的情况下更新拓扑代码或配置的重要策略。Storm 提供了多种机制来实现拓扑的滚动升级。

滚动升级的工作流程

准备新版本的 JAR 包
部署新代码到集群
逐步升级 Supervisor 节点
验证升级结果

上图展示了拓扑滚动升级的时间线，从准备新版本到最终完成升级的全过程。整个滚动升级过程可以分为准备阶段、执行阶段和验证阶段，每个阶段都有明确的操作步骤和时间点。

滚动升级实现方法

方法一：使用 Storm 原生滚动升级

# 1. 准备新版本 JAR
cp my-topology-new.jar /shared/storage/

# 2. 更新 storm.yaml 中的 JAR 路径
vim /opt/storm/conf/storm.yaml
# 添加或修改以下内容:
# topology.jar: "/shared/storage/my-topology-new.jar"

# 3. 重启 Supervisor
storm supervisor

# 4. 逐步升级拓扑
storm rebalance my-topology --wait 60  # 等待 60 秒让每个节点完成升级

# 5. 验证升级结果
storm list my-topology

方法二：使用自定义滚动升级脚本

#!/bin/bash

TOPOLOGY_NAME="my-topology"
NEW_JAR_PATH="/shared/storage/my-topology-new.jar"
NODES=("node1" "node2" "node3" "node4")
WAIT_TIME=60

# 1. 部署新 JAR
echo "Deploying new JAR to shared storage..."
cp $NEW_JAR_PATH /shared/storage/

# 2. 备份当前配置
echo "Backing up current configuration..."
cp /opt/storm/conf/storm.yaml /opt/storm/conf/storm.yaml.backup

# 3. 修改配置
echo "Updating storm.yaml..."
sed -i "s|topology.jar:.*|topology.jar: \"$NEW_JAR_PATH\"|" /opt/storm/conf/storm.yaml

# 4. 逐节点重启 Supervisor
for node in "${NODES[@]}"; do
    echo "Upgrading node: $node"
    ssh $node "sudo systemctl restart supervisor"
    sleep $WAIT_TIME
    echo "Verifying node: $node"
    ssh $node "jps | grep -q StormSubmitter"
done

# 5. 验证升级
echo "Verifying upgrade..."
storm list $TOPOLOGY_NAME

方法三：通过 Storm UI 进行滚动升级

登录 Storm Web UI
选择目标拓扑
点击 "Reconfig" 按钮
上传新的 JAR 包
调整配置参数
提交升级请求

滚动升级注意事项

版本兼容性：确保新旧版本间的数据格式兼容
状态迁移：有状态组件升级需要特别处理状态迁移
回滚计划：提前准备好回滚方案
监控告警：升级过程中加强监控，及时发现异常
业务影响：评估升级对业务的影响，必要时选择低峰期执行

实战示例与注意事项

下面是一个完整的动态扩缩容实践示例，包括 Rebalance、并行度调整和滚动升级的组合使用。

完整示例代码

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

public class DynamicScalingTopology {

    public static class SampleSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;
        private int count = 0;

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

        @Override
        public void nextTuple() {
            Utils.sleep(100);
            collector.emit(new Values("message-" + count++));
        }

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

    public static class SampleBolt extends BaseRichBolt {
        private OutputCollector collector;

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

        @Override
        public void execute(Tuple input) {
            String message = input.getString(0);
            System.out.println("Processing: " + message);
            // 模拟处理延迟
            Utils.sleep(50);
            collector.ack(input);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // 无输出
        }
    }

    public static StormTopology buildTopology(int spoutParallelism, int boltParallelism) {
        TopologyBuilder builder = new TopologyBuilder();
        
        // 设置初始并行度
        builder.setSpout("sample-spout", new SampleSpout(), spoutParallelism);
        builder.setBolt("sample-bolt", new SampleBolt(), boltParallelism)
               .shuffleGrouping("sample-spout");
                
        return builder.createTopology();
    }

    public static void main(String[] args) throws Exception {
        Config conf = new Config();
        conf.setNumWorkers(2);
        
        if (args != null && args.length > 0) {
            // 生产环境运行
            int spoutParallelism = Integer.parseInt(args[0]);
            int boltParallelism = Integer.parseInt(args[1]);
            
            StormTopology topology = buildTopology(spoutParallelism, boltParallelism);
            StormSubmitter.submitTopology("dynamic-scaling-topology", conf, topology);
        } else {
            // 本地测试
            int spoutParallelism = 2;
            int boltParallelism = 3;
            
            StormTopology topology = buildTopology(spoutParallelism, boltParallelism);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("dynamic-scaling-topology", conf, topology);
            
            // 模拟扩缩容
            Thread.sleep(10000);
            System.out.println("Increasing parallelism...");
            cluster.rebalance("dynamic-scaling-topology", 30000, 
                             new TopologyConfig().setComponentParallelism("sample-spout", 4)
                                               .setComponentParallelism("sample-bolt", 5));
            
            Thread.sleep(30000);
            System.out.println("Decreasing parallelism...");
            cluster.rebalance("dynamic-scaling-topology", 30000,
                             new TopologyConfig().setComponentParallelism("sample-spout", 2)
                                               .setComponentParallelism("sample-bolt", 3));
            
            Thread.sleep(30000);
            cluster.shutdown();
        }
    }
}

最小运行示例

编译打包:

mvn clean package -DskipTests

启动拓扑:

storm jar target/dynamic-scaling.jar DynamicScalingTopology 2 3

动态扩容:

storm rebalance dynamic-scaling-topology -e sample-spout:4,sample-bolt:5 -w 30

动态缩容:

storm rebalance dynamic-scaling-topology -e sample-spout:2,sample-bolt:3 -w 30

注意事项

数据一致性：扩缩容过程中可能出现数据重复或丢失，需要设计适当的去重或容错机制
资源争用：调整并行度时注意系统资源争用问题，避免过度占用导致集群不稳定
监控告警：建立完善的监控体系，及时发现扩缩容过程中的异常情况
渐进式调整：避免一次性大幅调整并行度，采用渐进式调整策略
业务影响评估：评估扩缩容对业务的影响，选择合适的调整时机

通过合理运用 Rebalance、在线调整并行度和拓扑滚动升级等技术，可以有效地实现 Storm 集群的动态扩缩容，提高资源利用率和系统稳定性，满足不同业务场景下的需求变化。

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

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

Apache Storm 动态扩缩容：Rebalance、在线调整并行度与拓扑滚动升级实战