阅读 this 和 this 后,我很难理解如何配置我的三叉戟拓扑。
基本上,我的 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();
}
所以我得到了一个 spout 和一些带有 parallelismHint 的操作(filter、groupBy)。
我不了解确定最佳并行度提示,此外,如果我在我的代码中设置此值,它如何与 Storm 标准拓扑配置结合使用,例如
topology.max.task.parallelism
topology.workers
topology.acker.executors
提前致谢
最佳答案
有一个很好的 gist by mrflip here 试图概述如何调整 Storm /三叉戟拓扑。这应该会指导您选择参数(您在问题中建议的参数和您可能还没有想到的其他参数)。
关于apache-storm - 三叉戟拓扑中的并行配置( Storm ),我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/18945196/