喷口逻辑错误

Question

在这个问题上我需要你的帮助。我读到 spout 负责读取数据或准备在 Bolt 中处理。所以我在 spout 中写了一些代码来打开文件并逐行读取

class SimSpout(storm.Spout):
    # Not much to do here for such a basic spout
    def initialize(self, conf, context):
    ## Open the file with read only permit
        self.f = open('data.txt', 'r')
    ## Read the first line
        self._conf = conf
        self._context = context
        storm.logInfo("Spout instance starting...")
    # Process the next tuple
    def nextTuple(self):
        # check if it reach at the EOF to close it
      for line in self.f.readlines():
        # Emit a random sentence
        storm.logInfo("Emiting %s" % line)
        storm.emit([line])

# Start the spout when it's invoked
SimSpout().run()

对吗？

Answer 1

您正在编写 Spout，它在 Storm 中的职责是发出元组以供下游螺栓处理。

Spout 的 nextTuple 职责是在每次调用它时发出一个事件。在您的代码中，您将发出文件中的所有行。如果您的单个元组是单行。您应该在文件中保留一个偏移量并阅读它 偏移线和发射，更新偏移量 = 偏移量 + 1。 像下面的东西

class SimSpout(storm.Spout):

  # Not much to do here for such a basic spout
  def initialize(self, conf, context):
    ## Open the file with read only permit
    self.f = open('data.txt', 'r')
    ## Read the first line
    self._conf = conf
    self._context = context
    self._offset = 0
    storm.logInfo("Spout instance starting...")

 # Process the next tuple
 def nextTuple(self):
    # check if it reach at the EOF to close it
    with open(self.f) as f:
      f.readlines()[self._offset]
      #Emit a random sentence
      storm.logInfo("Emiting %s" % line)
      storm.emit([line])
    self._offset = self._offset + 1