Spout of Storm流处理中关于nextTuple方法的问题答案

【问题标题】：Questions about the nextTuple method in the Spout of Storm stream processingSpout of Storm流处理中关于nextTuple方法的问题
【发布时间】：2014-03-05 20:24:28
【问题描述】：

我正在开发一些基于 Storm 的数据分析算法，对 Storm 的内部设计有一些疑问。我想在 Storm 中模拟传感器数据的产生和处理，因此我使用 Spout 通过在 Spout 的 nextTuple 方法中设置 sleep 方法，以恒定的时间间隔将传感器数据推送到后续的螺栓中。但是从实验结果来看，spout 似乎并没有以指定的速率推送数据。在实验中，系统中没有瓶颈螺栓。

然后我查看了一些关于 Storm 的 ack 和 nextTuple 方法的资料。现在我的疑问是 nextTuple 方法是否只有在前面的元组被完全处理并在 ack 方法中被确认时才被调用？

如果是这样，是否意味着我无法设置固定的时间间隔来发出数据？

非常感谢！

【问题讨论】：

我喜欢这个问题，但如果您在storm-users 邮件列表中提问，可能会更好。但是，我可以在下面看到我的回答，以获得一些我可以给你的信息。

标签： stream apache-storm

【解决方案1】：

我的经验是，您不应期望 Storm 做出任何实时保证，包括在您的情况下元组处理的速率。你当然可以编写一个只在某个时间安排上发出元组的 spout，但 Storm 并不能真正保证它总是会像你希望的那样频繁地调用 spout。

请注意，只要拓扑中有更多待处理的元组可用空间，就应该调用 nextTuple。如果拓扑有空闲容量，我希望 Storm 会尝试用它可以得到的任何东西来填充它。

【讨论】：

【解决方案2】：

我有一个类似的用例，我完成它的方法是使用TICK_TUPLE

Config tickConfig = new Config();                                       
tickConfig.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 15);
...
...
builder.setBolt("storage_bolt", new S3Bolt(), 4).fieldsGrouping("shuffle_bolt", new Fields("hash")).addConfigurations(tickConfig);

然后在我的storage_bolt 中（注意它是用python 写的，但你会明白的）我检查消息是否为tick_tuple，如果是则执行我的代码：

def process(self, tup): 
    if tup.stream == '__tick':                                              
        # Your logic that need to be executed every 15 seconds,
        # or what ever you specified in tickConfig.
        # NOTE: the maximum time is 600 s.                                            
        storm.ack(tup)                                                      
        return

【讨论】：