阿帕奇Storm三叉戟与Kafka喷口集成

Question

我无法找到正确集成Kafka和Apache Storm Trident的好文档。我试图查看相关的问题之前张贴在这里，但没有充分的信息。

GlobalPartitionInformation globalPartitionInformation  = new GlobalPartitionInformation(properties.getProperty("topic", "mytopic"));
Broker brokerForPartition0 = new Broker("IP1",9092);
Broker brokerForPartition1 = new Broker("IP2", 9092);
Broker brokerForPartition2 = new Broker("IP3:9092");

globalPartitionInformation.addPartition(0, brokerForPartition0);//mapping from partition 0 to brokerForPartition0
globalPartitionInformation.addPartition(1, brokerForPartition1);//mapping from partition 1 to brokerForPartition1
globalPartitionInformation.addPartition(2, brokerForPartition2);//mapping from partition 2 to brokerForPartition2
StaticHosts staticHosts = new StaticHosts(globalPartitionInformation);
TridentKafkaConfig tridentKafkaConfig = new TridentKafkaConfig(hosts,properties.getProperty("topic", "mytopic"));
tridentKafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
OpaqueTridentKafkaSpout kafkaSpout = new OpaqueTridentKafkaSpout(tridentKafkaConfig);

这样，我就可以为我的拓扑生成流，如下面的代码所示

TridentTopology topology = new TridentTopology();
Stream analyticsStream  = topology.newStream("spout", kafkaSpout).parallelismHint(Integer.valueOf(properties.getProperty("spout","6")))

虽然我提供了并行性和我的分区，但是只有一个Kafka Spout的执行器在运行，因此我无法很好地扩展它。

有谁能指导我更好地将Apache Storm Trident(2.0.0)与Apache Kafka(1.0)集成，每个集成3个节点集群吗？

2018-04-09 14:17:34.119 o.a.s.k.KafkaUtils Thread-15-spout-spout-executor[79 79] [INFO] Metrics Tick: Not enough data to calculate spout lag.  2018-04-09 14:17:34.129 o.a.s.k.KafkaUtils Thread-21-spout-spout-executor[88 88] [INFO] Metrics Tick: Not enough data to calculate spout lag.

Answer 1

如果你是在Storm2.0.0，我认为你应该切换到Storm-Kafka-客户端三叉戟喷口。storm-kafka模块只打算支持较旧的Kafka版本，因为底层的Kafka API（SimpleConsumer）正在被删除。新模块支持Kafka from 0.10.0.0和Forward。

您可以在这里找到新喷管的示例Trident拓扑https://github.com/apache/storm/blob/master/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/Trident/tridentkafkaclienttopologynamedtopics.java。