Storm+Kafka并行化不及预期

Question

负载的消息数量很少，每天大约2000条，但是每个任务都需要相当长的时间。特别是一个拓扑处理每个任务所需的时间是可变的，通常在1到20分钟之间。如果按顺序处理，吞吐量不足以处理所有传入消息。所有的拓扑和Kafka系统都安装在一台单机中（16个核心，16 GB的RAM）。

由于消息是独立的，并且可以并行处理，我们正在尝试使用Storm并发能力来提高吞吐量。

为此,拓扑配置如下：

null

null

private static StormTopology buildTopologyOD() {
    //This is the marker interface BrokerHosts.
    BrokerHosts hosts = new ZkHosts(configuration.getProperty(ZKHOSTS));
    TridentKafkaConfig tridentConfigCorrelation = new TridentKafkaConfig(hosts, configuration.getProperty(TOPIC_FROM_CORRELATOR_NAME));

    tridentConfigCorrelation.scheme = new RawMultiScheme();
    tridentConfigCorrelation.fetchSizeBytes = Integer.parseInt(configuration.getProperty(MAX_SIZE_BYTES_CORRELATED_STREAM));

    OpaqueTridentKafkaSpout spoutCorrelator = new OpaqueTridentKafkaSpout(tridentConfigCorrelation);

    TridentTopology topology = new TridentTopology();

    Stream existingObject = topology.newStream("kafka_spout_od1", spoutCorrelator)
            .shuffle()
            .each(new Fields("bytes"), new ProcessTask(), new Fields(RESULT_FIELD, OBJECT_FIELD))
            .parallelismHint(Integer.parseInt(configuration.getProperty(PARALLELISM_HINT)));

    //Create a state Factory to produce outputs to kafka topics.
    TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
            .withProducerProperties(kafkaProperties)
            .withKafkaTopicSelector(new ODTopicSelector())
            .withTridentTupleToKafkaMapper(new ODTupleToKafkaMapper());

    existingObject.partitionPersist(stateFactory, new Fields(RESULT_FIELD, OBJECT_FIELD), new TridentKafkaUpdater(), new Fields(OBJECT_FIELD));

    return topology.build();
}

和配置创建为：

private static Config createConfig(boolean local) {
    Config conf = new Config();
    conf.setMaxSpoutPending(1); // Also tried 2..6
    conf.setNumWorkers(4);

    return conf;
}

我们能做些什么来提高性能，或者通过增加并行任务的数量，或者在完成一批处理时避免饥饿？

Answer 1

我在storm-users上找到了一篇Nathan Marz关于为Trident设置并行度的老帖子：

我建议使用“name”函数来命名流的各个部分，这样UI就可以显示什么螺栓对应什么部分。

“三叉戟”将操作打包成尽可能少的螺栓。此外，它从不重新分区您的流，除非您已经执行了明确涉及重新分区的操作（例如，shuffle、groupBy、partitionBy、全局聚合等）。Trident的这个属性确保了您可以控制事情如何处理的排序/半排序。因此在这种情况下，groupBy之前的所有内容都必须具有相同的并行性，否则Trident将不得不重新划分流。因为您没有说要重新分区流，所以它不能这样做。通过引入一个重新分区操作，您可以为spout获得不同于每个spout的并行性，如下所示：