我编写了一个简单的 MapReduce 流程，用于从 Google Cloud Storage 上的文件中的 CSV 文件中读取行，然后创建一个实体。但是，我似乎无法让它在多个分片上运行。

代码使用了 mapreduce.control.start_map，看起来像这样。

class LoadEntitiesPipeline(webapp2.RequestHandler):
        id = control.start_map(map_name,
                          handler_spec="backend.line_processor",
                          reader_spec="mapreduce.input_readers.FileInputReader",
                          queue_name=get_queue_name("q-1"),
                          shard_count=shard_count,
                          mapper_parameters={
                              'shard_count': shard_count,
                              'batch_size': 50,
                              'processing_rate': 1000000,
                              'files': [gsfile],
                              'format': 'lines'})

我在两个地方都有 shard_count，因为我不确定哪些方法实际需要它。将 shard_count 设置为 8 到 32 之间的任何值都不会改变任何内容，因为状态页面始终显示 1/1 分片正在运行。为了分离事物，我让所有事物都在具有大量实例的后端队列上运行。我试过调整队列参数 per this wiki .最后，它似乎只是串行运行。

有什么想法吗？谢谢!

更新(还是没有成功):

在尝试隔离事物时，我尝试使用直接调用管道进行调用，如下所示:

class ImportHandler(webapp2.RequestHandler):

    def get(self, gsfile):
        pipeline = LoadEntitiesPipeline2(gsfile)
        pipeline.start(queue_name=get_queue_name("q-1"))

        self.redirect(pipeline.base_path + "/status?root=" + pipeline.pipeline_id)


class LoadEntitiesPipeline2(base_handler.PipelineBase):

    def run(self, gsfile):
        yield mapreduce_pipeline.MapperPipeline(
           'loadentities2_' + gsfile,
           'backend.line_processor',
           'mapreduce.input_readers.FileInputReader',
           params={'files': [gsfile], 'format': 'lines'},
           shards=32
        )

有了这个新代码，它仍然只在一个分片上运行。 我开始怀疑 mapreduce.input_readers.FileInputReader 是否能够按行并行化输入。

最佳答案

看起来 FileInputReader 只能通过文件进行分片。 format 参数只改变映射器函数被调用的方式。如果您将多个文件传递给映射器，它将开始在多个分片上运行。否则它只会使用一个分片来处理数据。

编辑#1:

深入挖掘 mapreduce 库之后。 MapReduce 将根据它定义的每个文件类型返回的 can_split 方法来决定是否将文件拆分成多个部分。目前，实现split 方法的唯一格式是ZipFormat。因此，如果您的文件格式不是 zip，它不会拆分文件以在多个分片上运行。

@classmethod
  def can_split(cls):
    """Indicates whether this format support splitting within a file boundary.

    Returns:
      True if a FileFormat allows its inputs to be splitted into
    different shards.
    """

https://code.google.com/p/appengine-mapreduce/source/browse/trunk/python/src/mapreduce/file_formats.py

但看起来可以编写自己的文件格式拆分方法。您可以先尝试破解并在 _TextFormat 上添加 split 方法，然后查看是否有多个分片在运行。

@classmethod
def split(cls, desired_size, start_index, opened_file, cache):
    pass

编辑#2:

一个简单的解决方法是让 FileInputReader 串行运行，但将耗时任务移动到并行 reduce 阶段。

def line_processor(line):
    # serial
    yield (random.randrange(1000), line)

def reducer(key, values):
    # parallel
    entities = []
    for v in values:
        entities.append(CREATE_ENTITY_FROM_VALUE(v))
    db.put(entities)

编辑#3:

如果尝试修改FileFormat，这里有一个例子(还没有测试过)

from file_formats import _TextFormat, FORMATS


class _LinesSplitFormat(_TextFormat):
  """Read file line by line."""

  NAME = 'split_lines'

  def get_next(self):
    """Inherited."""
    index = self.get_index()
    cache = self.get_cache()
    offset = sum(cache['infolist'][:index])

    self.get_current_file.seek(offset)
    result = self.get_current_file().readline()
    if not result:
      raise EOFError()
    if 'encoding' in self._kwargs:
      result = result.encode(self._kwargs['encoding'])
    return result

  @classmethod
  def can_split(cls):
    """Inherited."""
    return True

  @classmethod
  def split(cls, desired_size, start_index, opened_file, cache):
    """Inherited."""
    if 'infolist' in cache:
      infolist = cache['infolist']
    else:
      infolist = []
      for i in opened_file:
        infolist.append(len(i))
        cache['infolist'] = infolist

    index = start_index
    while desired_size > 0 and index < len(infolist):
      desired_size -= infolist[index]
      index += 1
    return desired_size, index


FORMATS['split_lines'] = _LinesSplitFormat

然后可以通过将 mapper_parameters 从 lines 更改为 split_line 来调用新的文件格式。

class LoadEntitiesPipeline(webapp2.RequestHandler):
    id = control.start_map(map_name,
                      handler_spec="backend.line_processor",
                      reader_spec="mapreduce.input_readers.FileInputReader",
                      queue_name=get_queue_name("q-1"),
                      shard_count=shard_count,
                      mapper_parameters={
                          'shard_count': shard_count,
                          'batch_size': 50,
                          'processing_rate': 1000000,
                          'files': [gsfile],
                          'format': 'split_lines'})

关于python - 如何让 AppEngine map reduce 横向扩展？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/19195215/