【问题标题】:parallelism configuration in trident topology (storm)三叉戟拓扑中的并行配置(风暴)
【发布时间】:2013-09-27 12:28:27
【问题描述】:

阅读thisthis 后,我很难理解如何配置我的三叉戟拓扑。

基本上,我的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、grouopBy)。 我不知道确定最佳并行度提示,此外,如果我在我的代码中设置此值,它如何与 Storm 标准拓扑配置(例如

)结合使用
topology.max.task.parallelism
topology.workers
topology.acker.executors

提前致谢

【问题讨论】:

    标签: apache-storm apache-kafka trident


    【解决方案1】:

    有一个优秀的gist by mrflip here 试图概述如何调整风暴/三叉戟拓扑。这应该会指导您选择参数(包括您在问题中建议的参数以及您可能尚未想到的其他参数)。

    【讨论】:

    • 一见钟情。我会深入研究并更新,谢谢
    • 我无法打开该页面,您能否再给我一个可访问的页面?非常感谢。
    • 嗨@Yohn 它应该可以公开访问。
    • 好的,我明白了。我的局域网一定有问题。还是谢谢你。
    猜你喜欢
    • 2013-08-06
    • 1970-01-01
    • 2017-05-28
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2015-04-14
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多