如何将参数动态传递给 GAE mapReduce 上的映射函数

Question

我需要运行一个动态的mapreduce作业，因为每次运行mapreduce作业时都需要将参数传递给map和reduce函数（例如，响应用户请求）。

我该如何实现此目的？我在文档中看不到如何在运行时对映射和减少进行动态处理。

class MatchProcessing(webapp2.RequestHandler):
  def get(self):
      requestKeyID=int(self.request.get('riderbeeRequestID'))
      userKey=self.request.get('userKey')
      pipeline = MatchingPipeline(requestKeyID, userKey)
      pipeline.start()
      self.redirect(pipeline.base_path + "/status?root=" + pipeline.pipeline_id)

class MatchingPipeline(base_handler.PipelineBase):
    def run(self, requestKeyID, userKey):
        yield mapreduce_pipeline.MapreducePipeline(
            "riderbee_matching",
            "tasks.matchingMR.riderbee_map",
            "tasks.matchingMR.riderbee_reduce",
            "mapreduce.input_readers.DatastoreInputReader",
            "mapreduce.output_writers.BlobstoreOutputWriter",
            mapper_params={
                "entity_kind": "models.rides.RiderbeeRequest",
                "requestKeyID": requestKeyID,
                "userKey": userKey,
            },
            reducer_params={
                "mime_type": "text/plain",
            },
            shards=16)

def riderbee_map(riderbeeRequest):
    # would like to access the requestKeyID and userKey parameters that were passed in mapper_params
    # so that we can do some processing based on that
    yield (riderbeeRequest.user.email, riderbeeRequest.key().id())

def riderbee_reduce(key, values):
    # would like to access the requestKeyID and userKey parameters that were passed earlier, perhaps through reducer_params
    # so that we can do some processing based on that
    yield "%s: %sn" % (key, len(values))

请帮忙？

Answer 1

我很确定您可以在mapper_parameters中指定参数，然后从上下文模块中读取它们。有关更多详细信息，请参阅 http://code.google.com/p/appengine-mapreduce/wiki/UserGuidePython#Mapper_parameters。

Answer 2

以下是使用上下文模块从映射器函数访问映射器参数的方法：

from mapreduce import context
def riderbee_map(riderbeeRequest):
    ctx = context.get()
    params = ctx.mapreduce_spec.mapper.params
    requestKeyID = params["requestKeyID"]