如何将多个（不同的）元组从一个KafkaSpout一次发送到bolt？

Question

我是阿帕奇Storm的新手。

请救命！！

我的拓扑：

TopologyBuilder builder=new TopologyBuilder（）；

这里，我试图从事件流中获取长度为3的窗口中的第一个和最后一个事件。但是我得到的第一个和最后一个事件是一样的，因为KafkaSpout一次只发送一个元组。

Answer 1

spout不能这样做，但是您可以使用Storm的窗口支持https://storm.apache.org/releases/2.0.0-snapshot/windowing.html，或者只编写一个聚合螺栓并将其放在spout和拓扑的其余部分之间。

所以您的拓扑结构应该是spout->aggregator->feature selection->trend detection。

我建议您尝试一下内置的窗口化支持，但是如果您更愿意编写自己的聚合，那么您的bolt实际上只需要接收一些元组（例如3)，并发出一个包含所有值的新元组。

聚合器螺栓应该执行如下操作

private List<Tuple> buffered;

execute(Tuple input) {
  if (buffered.size != 2) {
    buffered.add(input)
    return
  }
  Tuple first = buffered.get(0)
  Tuple second = buffered.get(1)
  Values aggregate = new Values(first.getValues(), second.getValues(), input.getValues())
  List<Tuple> anchors = List.of(first, second, input)
  collector.emit(anchors, aggregate)
  collector.ack(first, second, input)
  buffered.clear()
}

这样，您就得到一个包含3个输入元组内容的元组。