trident拓扑中的并行配置(storm)

unguejic  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(295)

在阅读了这个和这个之后,我很难理解如何配置我的三叉戟拓扑。
基本上,我的storm应用程序是从Kafka那里读取数据,进行一些数据操作,最后向Cassandra写信。
以下是我当前构建拓扑的方式:

private static StormTopology buildTopology() {
// connection to kafka
ZkHosts zkHosts = new ZkHosts(broker_zk, broker_path);
TridentKafkaConfig kafkaConfig = new TridentKafkaConfig(zkHosts, topic);
kafkaConfig.scheme = new RawMultiScheme();
StateFactoryFields[] cassandraStateFactories = createStateFactories();
TransactionalTridentKafkaSpout spout = new TransactionalTridentKafkaSpout(kafkaConfig);
TridentTopology topology = new TridentTopology();
Stream kafkaSpout = topology.newStream("kafkaspout", spout).parallelismHint(1).shuffle();
Stream filterValidatStream = kafkaSpout.each(new Fields("bytes"), new SplitKafkaInput(), EventData.getEventDataFields()).parallelismHint(1);
for (StateFactoryFields stateFactoryFields : cassandraStateFactories) {
    filterValidatStream.groupBy(stateFactoryFields.groupingFields)
        .persistentAggregate(stateFactoryFields.cassandraStateFactor, new Count(), new Fields("count")).parallelismHint(2);
}
logger.info("Building topology");
return topology.build();
}

因此,我得到了一个喷口和一些操作(过滤器,groupby)与parallelishmint。我不知道如何确定最佳的parallelishmint,而且如果我在代码中设置这个值,它如何与标准的storm拓扑配置(如

topology.max.task.parallelism
topology.workers
topology.acker.executors

提前谢谢

dfddblmv

dfddblmv1#

mrflip在这里提供了一个很好的要点,试图概述如何调整storm/trident拓扑结构。这将指导您选择参数(包括您在问题中建议的参数和您可能尚未想到的其他参数)。

相关问题