如何使用 Python API 开发 Storm 拓扑？

推荐答案

// www.javascriptcn.com code example
from storm import TopologyBuilder, Spout, Bolt, emit, log

class SentenceSpout(Spout):
    def initialize(self, conf, context):
        self.sentences = [
            "storm is a distributed realtime computation system",
            "it processes streams of data",
            "and provides high throughput and low latency"
        ]
        self.index = 0

    def nextTuple(self):
        if self.index < len(self.sentences):
            emit([self.sentences[self.index]])
            self.index += 1
        else:
            self.index = 0

class SplitSentenceBolt(Bolt):
    def process(self, tup):
        sentence = tup.values[0]
        for word in sentence.split():
            emit([word])

class WordCountBolt(Bolt):
    def initialize(self, conf, context):
        self.counts = {}

    def process(self, tup):
        word = tup.values[0]
        self.counts[word] = self.counts.get(word, 0) + 1
        log(f"Word: {word}, Count: {self.counts[word]}")
        emit([word, self.counts[word]])

def create_topology():
    builder = TopologyBuilder()
    builder.setSpout("sentence-spout", SentenceSpout())
    builder.setBolt("split-sentence-bolt", SplitSentenceBolt()).shuffleGrouping("sentence-spout")
    builder.setBolt("word-count-bolt", WordCountBolt()).fieldsGrouping("split-sentence-bolt", ["word"])
    return builder.create_topology()

if __name__ == "__main__":
    from storm import LocalCluster
    cluster = LocalCluster()
    cluster.submit_topology("word-count-topology", create_topology())
    cluster.shutdown()

本题详细解读

1. 拓扑结构

Storm 拓扑由 Spout 和 Bolt 组成。Spout 是数据源，负责从外部数据源读取数据并发送到拓扑中。Bolt 是处理单元，负责对数据进行处理、转换或聚合。

2. Spout 实现

SentenceSpout 是一个简单的 Spout，它从一个预定义的句子列表中逐条发送数据。initialize 方法用于初始化 Spout，nextTuple 方法用于发送数据。

3. Bolt 实现

• SplitSentenceBolt 负责将句子拆分为单词。process 方法接收来自 Spout 的元组，将其拆分为单词并发送出去。
• WordCountBolt 负责统计单词出现的次数。initialize 方法用于初始化计数器，process 方法用于更新计数器并输出结果。

4. 拓扑构建

create_topology 函数使用 TopologyBuilder 构建拓扑。setSpout 和 setBolt 方法分别用于设置 Spout 和 Bolt，shuffleGrouping 和 fieldsGrouping 方法用于定义数据流的分组方式。

5. 本地集群运行

LocalCluster 用于在本地运行 Storm 拓扑。submit_topology 方法提交拓扑，shutdown 方法用于关闭集群。

通过以上步骤，你可以使用 Python API 开发并运行一个简单的 Storm 拓扑。