如何创建和提交 Storm 拓扑？

推荐答案

创建 Storm 拓扑

1. 定义 Spout：Spout 是数据流的来源。你可以通过实现 IRichSpout 接口或继承 BaseRichSpout 类来定义 Spout。

   // www.javascriptcn.com code example
   public class MySpout extends BaseRichSpout {
       private SpoutOutputCollector collector;
   
       @Override
       public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
           this.collector = collector;
       }
   
       @Override
       public void nextTuple() {
           // 生成数据并发射
           collector.emit(new Values("data"));
       }
   
       @Override
       public void declareOutputFields(OutputFieldsDeclarer declarer) {
           declarer.declare(new Fields("field"));
       }
   }

2. 定义 Bolt：Bolt 是数据流的处理单元。你可以通过实现 IRichBolt 接口或继承 BaseRichBolt 类来定义 Bolt。

   // www.javascriptcn.com code example
   public class MyBolt 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 data = input.getStringByField("field");
           // 发射处理后的数据
           collector.emit(new Values(data));
       }
   
       @Override
       public void declareOutputFields(OutputFieldsDeclarer declarer) {
           declarer.declare(new Fields("processed_field"));
       }
   }

3. 构建拓扑：使用 TopologyBuilder 类将 Spout 和 Bolt 组合成一个拓扑。

   TopologyBuilder builder = new TopologyBuilder();
   builder.setSpout("mySpout", new MySpout());
   builder.setBolt("myBolt", new MyBolt()).shuffleGrouping("mySpout");

提交 Storm 拓扑

1. 配置拓扑：创建一个 Config 对象来配置拓扑。

   Config config = new Config();
   config.setDebug(true);

2. 提交拓扑：使用 StormSubmitter 类将拓扑提交到 Storm 集群。

   StormSubmitter.submitTopology("myTopology", config, builder.createTopology());

本题详细解读

创建 Storm 拓扑

在 Storm 中，拓扑是由 Spout 和 Bolt 组成的有向无环图（DAG）。Spout 负责从外部数据源读取数据并发射到拓扑中，而 Bolt 则负责处理这些数据并可能将结果发射到其他 Bolt 或外部存储。

1. Spout：Spout 是拓扑的源头，负责从外部数据源（如 Kafka、数据库等）读取数据并将其发射到拓扑中。nextTuple() 方法是 Spout 的核心方法，Storm 会不断调用该方法来获取数据。

2. Bolt：Bolt 是拓扑中的处理单元，负责接收来自 Spout 或其他 Bolt 的数据并进行处理。execute() 方法是 Bolt 的核心方法，Storm 会调用该方法来处理传入的 Tuple。

3. 拓扑构建：使用 TopologyBuilder 类可以将 Spout 和 Bolt 组合成一个拓扑。shuffleGrouping() 方法用于指定 Bolt 如何从 Spout 或其他 Bolt 接收数据。

提交 Storm 拓扑

1. 配置拓扑：Config 对象用于配置拓扑的各种参数，如调试模式、工作进程数等。

2. 提交拓扑：StormSubmitter.submitTopology() 方法用于将拓扑提交到 Storm 集群。提交后，Storm 会启动拓扑并在集群中运行。

通过以上步骤，你可以成功创建并提交一个 Storm 拓扑。