python - 在 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) 的 session 描述中写道“现在可以在 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 中实现这个……视频显示了具有定义的映射和化简函数的示例，但它们似乎是与数据存储无关的非常通用的示例。我发现的所有其他示例都使用一个函数来处理来自 DatastoreInputReader 的数据，但它们似乎只是“映射”阶段，没有示例说明如何进行“减少”(以及如何将减少结果存储在数据存储)。

Answer 1

您实际上并不需要 reduce 阶段。您可以使用线性任务链来完成此操作，大致如下所示:

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)

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

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

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