【问题标题】：Files processed in mapreduce在 mapreduce 中处理的文件
【发布时间】：2014-03-17 00:25:28
【问题描述】：

我正在尝试确定 AppEngine 上的 mapreduce 作业处理了哪些文件。我在 Cloud Storage 存储分区中使用通配符：

class FilesPipeline(base_handler.PipelineBase):

    def run(self):

        output_blobstore_ids = yield mapreduce_pipeline.MapreducePipeline(
            "proc_files",
            "project.task.proc_files.mapper",
            "project.task.proc_files.reducer",
            "mapreduce.input_readers.FileInputReader",
            "mapreduce.output_writers.BlobstoreOutputWriter"
            ,mapper_params={
              'shard_count': 4,
              'batch_size': 50,
              'files': ['/gs/project_inbox/partner*'],
              'format': 'lines'}
            ,reducer_params={
            },
            shards=4)

        yield StoreOutput(output_blobstore_ids)

理想情况下，我想在 StoreOutput 实例中获取由 mapreduce 管道处理的文件列表，但任何地方都可以。

谢谢！

【问题讨论】：

标签： python google-app-engine mapreduce

【解决方案1】：

这是我发现的：

准备管道

files_list=self.get_files_list()
pipeline=FilesPipeline(mapper_files=files_list)
pipeline.start(queue_name='proc-files')

管道定义

类 FilesPipeline(base_handler.PipelineBase): def run(self, mapper_files=[]):

    output_blobstore_ids = yield mapreduce_pipeline.MapreducePipeline(
        "proc_files",
        "project.task.proc_files.mapper",
        "project.task.proc_files.reducer",
        "mapreduce.input_readers.FileInputReader",
        "mapreduce.output_writers.BlobstoreOutputWriter"
        ,mapper_params={
          'shard_count': 4
          ,'batch_size': 50
          ,"files":      mapper_files
          ,'format':     'lines'}
        ,reducer_params={
        },
        shards=4)

    yield StoreOutput(output_blobstore_ids)

希望这对某人有所帮助。

【讨论】：