Storm ShellBolt“丢失”元组

Question

我有一个拓扑结构如下所示：

注意，Bolt2和Bolt3从Bolt1和Bolt4接收元组。所有bolt都是运行python脚本的ShellBolts，而spout是运行从RabbitMQ读取的python脚本的ShellSpout。除了Bolt4之外，所有的工作都是预期的。如果我在RabbitMQ中一次添加一条消息，它就会全部工作并且干净利落地完成。如果我在Bolt4上挂起消息时对消息进行排队，则Bolt4将永远不会处理该消息。其他螺栓仍然执行它们的功能，但Bolt4将只是坐在那里完成第一个。

Storm UI显示所有的元组都由bolt4执行，但只有一个被加密。没有失败。我使用的是Storm0.9.5和Storm-Starter中包含的multilang python适配器。

编辑：这里有一个最小的失败案例。

螺栓1.py:

import storm
import time
import json


class Bolt1(storm.BasicBolt):
    # overrides storm.Bolt.process

    def process(self, tup):
        objID, APIArgs = tup.values
        APIArgs = json.loads(APIArgs)
        self.emit("bolt3Queue", objID, **APIArgs)
        self.emit("bolt2Queue", objID, **APIArgs)
        self.emit("bolt4Queue", objID, **APIArgs)
        storm.ack(tup)

    def emit(self, stream, objID, **APIArgs):
        tup = [objID, json.dumps(APIArgs)]
        storm.log("Emit [%s] %s" % (stream, tup))
        storm.emit(tup, stream=stream)

if __name__ == '__main__':
    Bolt1().run()

螺栓2.py:

import storm
import time
import json


class Bolt2(storm.BasicBolt):
    # overrides storm.Bolt.process

    def process(self, tup):
        objID, APIArgs = tup.values
        APIArgs = json.loads(APIArgs)
        storm.ack(tup)

    def emit(self, stream, objID, **APIArgs):
        tup = [objID, json.dumps(APIArgs)]
        storm.log("Emit [%s] %s" % (stream, tup))
        storm.emit(tup, stream=stream)

if __name__ == '__main__':
    Bolt2().run()

import storm
import time
import json


class Bolt3(storm.BasicBolt):
    # overrides storm.Bolt.process

    def process(self, tup):
        objID, APIArgs = tup.values
        APIArgs = json.loads(APIArgs)
        storm.ack(tup)

    def emit(self, stream, objID, **APIArgs):
        tup = [objID, json.dumps(APIArgs)]
        storm.log("Emit [%s] %s" % (stream, tup))
        storm.emit(tup, stream=stream)

if __name__ == '__main__':
    Bolt3().run()

import storm
import time
import json


class Bolt4(storm.BasicBolt):
    # overrides storm.Bolt.process

    def process(self, tup):
        objID, APIArgs = tup.values
        APIArgs = json.loads(APIArgs)
        self.emit("bolt3Queue", objID, **APIArgs)
        self.emit("bolt2Queue", objID, **APIArgs)
        storm.ack(tup)

    def emit(self, stream, objID, **APIArgs):
        tup = [objID, json.dumps(APIArgs)]
        storm.log("Emit [%s] %s" % (stream, tup))
        storm.emit(tup, stream=stream)

if __name__ == '__main__':
    Bolt4().run()

import storm
import random

class Spout(storm.Spout):

    def nextTuple(self):
        storm.emit(["id1234", "{}"], id=str(random.randint(1, 10000)))

if __name__ == '__main__':
    Spout().run()

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package storm.starter;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.ShellSpout;
import backtype.storm.task.ShellBolt;
import backtype.storm.topology.*;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;


public class PyroTopology {
  public static class PythonBolt extends ShellBolt implements IRichBolt {

    public PythonBolt(String script) {
      super("python", script);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {}

    @Override
    public Map<String, Object> getComponentConfiguration() {
      return null;
    }
  }

  public static class Bolt4 extends ShellBolt implements IRichBolt {

    public Bolt4() {
      super("python", "bolt4.py");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declareStream("bolt3Queue", new Fields("objID", "APIArgs"));
      declarer.declareStream("bolt2Queue", new Fields("objID", "APIArgs"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
      return null;
    }
  }

  public static class Bolt1 extends ShellBolt implements IRichBolt {

    public Bolt1() {
      super("python", "bolt1.py");
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declareStream("bolt3Queue", new Fields("objID", "APIArgs"));
      declarer.declareStream("bolt2Queue", new Fields("objID", "APIArgs"));
      declarer.declareStream("bolt4Queue", new Fields("objID", "APIArgs"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
      return null;
    }
  }

  public static class PythonSpout extends ShellSpout implements IRichSpout {

    public PythonSpout() {
      super("python", "spout.py");
    }

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

    @Override
    public Map<String, Object> getComponentConfiguration() {
      return null;
    }
  }

  public static void main(String[] args) throws Exception {

    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("rabbit", new PythonSpout(), 1);

    builder.setBolt("bolt1", new Bolt1(), 1).
            shuffleGrouping("rabbit");

    builder.setBolt("bolt4", new Bolt4(), 1).
            shuffleGrouping("bolt1", "bolt4Queue");

    builder.setBolt("bolt3", new PythonBolt("bolt3.py"), 1).
            shuffleGrouping("bolt1", "bolt3Queue").
            shuffleGrouping("bolt4", "bolt3Queue");

    builder.setBolt("bolt2", new PythonBolt("bolt2.py"), 1).
            shuffleGrouping("bolt1", "bolt2Queue").
            shuffleGrouping("bolt4", "bolt2Queue");

    Config conf = new Config();
    conf.setStatsSampleRate(1.0);
    conf.put(Config.TOPOLOGY_DEBUG, true);
    conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 5);
    conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 60);

    if (args != null && args.length > 0) {
      conf.setNumWorkers(3);

      StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
    }
    else {
      conf.setMaxTaskParallelism(3);

      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("word-count", conf, builder.createTopology());

      Thread.sleep(30000);

      cluster.shutdown();
    }
  }
}

storm jar target/storm-starter-0.9.5-jar-with-dependencies.jar storm.starter.PyroTopology vb

Answer 1

您正在使用basicbolt，它会为您自动处理加密。因此，不能在代码中手动ack输入元组。这会导致单个元组的多个ack，从而混淆了Storm跟踪ack的机制（通过对消息ID和ack ID进行异或）。作为另一种选择（如果您需要高级的加密行为，您可以实现bolt。

如UI所示，spout没有接收到ack，因此当达到最大spout挂起时，Storm会停止发出元组。此外，您会看到bolt的“executed”和“acked”的计数不匹配--这也表明没有正确处理acks。