在Google App Engine中使用mapreduce的简单计数器示例

Question

我对GAE中mapreduce支持的当前状态感到困惑.根据文档http://code.google.com/p/appengine-mapreduce/还不支持reduce阶段,但在I/O 2011会话描述中(http://www.youtube.com)/watch？v = EIxelKcyCC0)它写的"现在可以在App Engine上运行完整的Map Reduce作业".我想知道我是否可以在此任务中使用mapreduce:

我想做的事:

我有车型的车型:

class Car(db.Model):
    color = db.StringProperty()

我想运行mapreduce进程(不时地,cron定义),它可以计算每种颜色中有多少汽车,并将此结果存储在数据存储区中.看起来像一个非常适合mapreduce的工作(但是如果我错了更正),阶段"map"将为每个Car实体产生对(,1),而阶段"reduce"应该通过color_name合并这些数据给出我预期的结果.我想得到的最终结果是存储在数据存储区中的计算数据的实体,类似于:

class CarsByColor(db.Model):
    color_name = db.StringProperty()
    cars_num = db.IntegerProperty()

问题: 我不知道如何在appengine中实现这个...视频显示了定义map和reduce函数的示例,但它们似乎是与数据存储无关的非常一般的示例.我发现的所有其他示例都使用一个函数来处理来自DatastoreInputReader的数据,但它们似乎只是"map"阶段,没有关于如何执行"reduce"的示例(以及如何将结果存储在数据存储).

Answer 1

你真的不需要减少阶段.您可以使用线性任务链完成此操作,或多或少如下:

def count_colors(limit=100, totals={}, cursor=None):
  query = Car.all()
  if cursor:
    query.with_cursor(cursor)
  cars = query.fetch(limit)
  for car in cars:
    try:
      totals[car.color] += 1
    except KeyError:
      totals[car.color] = 1
  if len(cars) == limit:
    cursor = query.cursor()
    return deferred.defer(count_colors, limit, totals, cursor)
  entities = []
  for color in totals:
    entity = CarsByColor(key_name=color)
    entity.cars_num = totals[color]
    entities.append(entity)
  db.put(entities)

deferred.defer(count_colors)

这应该遍历所有汽车,将查询光标和正在运行的计数传递给一系列临时任务,并将总计存储在最后.

如果必须在单个模型中合并来自多个数据存储,多个模型或多个索引的数据,则减少阶段可能有意义.因为我不认为它会给你买任何东西.

另一种选择:使用任务队列来维护每种颜色的实时计数器.创建汽车时,启动任务以增加该颜色的总数.更新汽车时,启动一项任务以减少旧颜色,另一项任务增加新颜色.以事务方式更新计数器以避免竞争条件.

Answer 2

我在这里提供的解决方案我最终使用来自GAE的mapreduce(没有减少阶段).如果我从头开始,我可能会使用Drew Sears提供的解决方案.

它适用于GAE python 1.5.0

在app.yaml中我添加了mapreduce的处理程序:

- url: /mapreduce(/.*)?
  script: $PYTHON_LIB/google/appengine/ext/mapreduce/main.py

和我的mapreduce代码的处理程序(我使用url/mapred_update来收集mapreduce产生的结果):

- url: /mapred_.*
  script: mapred.py

创建了用于处理Car实体的mapreduce.yaml:

mapreduce:
- name: Color_Counter
  params:
  - name: done_callback
    value: /mapred_update
  mapper:
    input_reader: google.appengine.ext.mapreduce.input_readers.DatastoreInputReader
    handler: mapred.process
    params:
    - name: entity_kind
      default: models.Car

说明:done_callback是在mapreduce完成其操作后调用的url.mapred.process是一个处理单个实体和更新计数器的函数(它在mapred.py文件中定义).Model Car在models.py中定义

mapred.py:

from models import CarsByColor
from google.appengine.ext import db
from google.appengine.ext.mapreduce import operation as op
from google.appengine.ext.mapreduce.model import MapreduceState

from google.appengine.ext import webapp
from google.appengine.ext.webapp.util import run_wsgi_app

def process(entity):
    """Process individual Car"""
    color = entity.color
    if color:
        yield op.counters.Increment('car_color_%s' % color)

class UpdateCounters(webapp.RequestHandler):
    """Create stats models CarsByColor based on the data 
    gathered by mapreduce counters"""
    def post(self):
        """Called after mapreduce operation are finished"""
        # Finished mapreduce job id is passed in request headers
        job_id = self.request.headers['Mapreduce-Id']
        state = MapreduceState.get_by_job_id(job_id)
        to_put = []
        counters = state.counters_map.counters
        # Remove counter not needed for stats
        del counters['mapper_calls']
        for counter in counters.keys():
            stat = CarsByColor.get_by_key_name(counter)
            if not stat:
                stat = CarsByColor(key_name=counter,
                                name=counter)
            stat.value = counters[counter]
            to_put.append(stat)
        db.put(to_put)

        self.response.headers['Content-Type'] = 'text/plain'
        self.response.out.write('Updated.')


application = webapp.WSGIApplication(
                                     [('/mapred_update', UpdateCounters)],
                                     debug=True)
def main():
    run_wsgi_app(application)

if __name__ == "__main__":
    main()            

与问题相比,CarsByColor模型的定义略有变化.

你可以从url手动启动mapreduce作业:http:// yourapp/mapreduce /并希望来自cron(我还没有测试过cron).