链接MapReduces-Google AppEngine

Question

以下是管道代码：

class SongsPurchasedTogetherPipeline(base_handler.PipelineBase):

  def run(self, filekey, blobkey):
    bucket_name = app_identity.get_default_gcs_bucket_name()
    intermediate_output = yield mapreduce_pipeline.MapreducePipeline(
        "songs_purchased_together_intermediate",
        "main.songs_purchased_together_map1",
        "main.songs_purchased_together_reduce1",
        "mapreduce.input_readers.BlobstoreLineInputReader",
        "mapreduce.output_writers.GoogleCloudStorageOutputWriter",
        mapper_params={
            "blob_keys": blobkey,
        },
        reducer_params={
            "output_writer": {
                "bucket_name": bucket_name,
                "content_type": "text/plain",
            }
        },
        shards=1)
    yield StoreOutput("SongsPurchasedTogetherIntermediate", filekey, intermediate_output)

    intermediate_output_key = yield BlobKey(intermediate_output)
    output = yield mapreduce_pipeline.MapreducePipeline(
        "songs_purchased_together",
        "main.songs_purchased_together_map2",
        "main.songs_purchased_together_reduce2",
        "mapreduce.input_readers.BlobstoreLineInputReader",
        "mapreduce.output_writers.GoogleCloudStorageOutputWriter",
        mapper_params=(intermediate_output_key),
        reducer_params={
            "output_writer": {
                "bucket_name": bucket_name,
                "content_type": "text/plain",
            }
        },
        shards=1)
    yield StoreOutput("SongsPurchasedTogether", filekey, output)

下面是BlobKey类，它接受中间输出并生成供Map2使用的blob密钥：

class BlobKey(base_handler.PipelineBase):

  def run(self, output):
    blobstore_filename = "/gs" + output[0]
    blobstore_gs_key = blobstore.create_gs_key(blobstore_filename)
    return {
      "blob_keys": blobstore_gs_key
    }

StoreOutput类与Google的MapReduce demo https://github.com/googleCloudplatform/appengine-mapReduce/blob/master/python/demo/main.py中的类相同，并且执行与BlobKey类相同的操作，但还将Blob的URL作为链接发送到HTML。

Answer 1

好吧，我发现Google已经从GAE GitHub存储库的标准编写器列表中删除了BlobstoreOutputWriter，这使得事情变得更加复杂。我不得不写信给谷歌云商店，并从那里阅读。我编写了一个帮助器类，它为GoogleLoudStorageInputReader生成映射器参数。

class GCSMapperParams(base_html" target="_blank">handler.PipelineBase):

  def run(self, GCSPath):
    bucket_name = app_identity.get_default_gcs_bucket_name()
    return {
            "input_reader": {
                "bucket_name": bucket_name,
                "objects": [path.split('/', 2)[2] for path in GCSPath],
            }
        }

该函数将使用GoogleLoudStorageOutputWriter的一个MapReduce阶段的输出作为参数，并返回可以分配给下一个MapReduce阶段的mapper_params的字典。

基本上，第一个MapReduce阶段的输出值是一个包含 / /key/output-[i] 的列表，其中i是碎片的数量。为了使用GoogleCloudStorageInputReader，数据的键应该通过Mapper_params中的变量Objects传递。键的形式必须是key/output-[i]，因此helper类只需从其中移除 / / 。