在App Engine中使用MapReduce进行GROUP BY

Question

我想找一种方法，在数据存储中使用MapReduce进行GROUP BY操作。根据我所知，App Engine在GQL中不支持GROUP BY，而其他开发者建议的一个好方法是使用MapReduce。

我下载了源代码，并在研究示例代码，我尝试在我的案例中实现它。但我没有成功。这是我尝试的方法。也许我做的所有事情都是错的。如果有人能帮我做到这一点，我将非常感激。

我想做的是：我在数据存储中有一堆联系人，每个联系人都有一个日期。有很多重复的联系人，日期是一样的。我想做的就是简单地将相同日期的联系人分组在一起。

例如：

假设我有这些联系人：

联系人姓名：Foo1 | 日期：01-10-2012
联系人姓名：Foo2 | 日期：02-05-2012
联系人姓名：Foo1 | 日期：01-10-2012

所以在MapReduce操作之后，结果应该是这样的：

联系人姓名：Foo1 | 日期：01-10-2012
联系人姓名：Foo2 | 日期：02-05-2012

对于GROUP BY功能，我觉得单词计数可以完成这个工作。

编辑

日志中显示的唯一内容是：

/mapreduce/pipeline/run 200

运行 GetContactData.WordCountPipeline((u'2012-02-02',), *{})#da26a9b555e311e19b1e6d324d450c1a

编辑结束

如果我做错了什么，或者如果我用MapReduce做GROUP BY的方法不对，请帮我看看该如何用MapReduce来实现。

这是我的代码：

from Contacts import Contacts
from google.appengine.ext import webapp
from google.appengine.ext.webapp import template
from google.appengine.ext.webapp.util import run_wsgi_app
from google.appengine.api import mail
from google.appengine.ext.db import GqlQuery
from google.appengine.ext import db


from google.appengine.api import taskqueue
from google.appengine.api import users

from mapreduce.lib import files
from mapreduce import base_handler
from mapreduce import mapreduce_pipeline
from mapreduce import operation as op
from mapreduce import shuffler

import simplejson, logging, re


class GetContactData(webapp.RequestHandler):

    # Get the calls based on the user id
    def get(self):
        contactId = self.request.get('contactId')
        query_contacts = Contact.all()
        query_contacts.filter('contact_id =', int(contactId))
        query_contacts.order('-timestamp_')
        contact_data = []
        if query_contacts != None:
            for contact in query_contacts:
                    pipeline = WordCountPipeline(contact.date)
                    pipeline.start()
                    record = { "contact_id":contact.contact_id,
                               "contact_name":contact.contact_name,
                               "contact_number":contact.contact_number,
                               "timestamp":contact.timestamp_,
                               "current_time":contact.current_time_,
                               "type":contact.type_,
                               "current_date":contact.date }
                    contact_data.append(record)

        self.response.headers['Content-Type'] = 'application/json'
        self.response.out.write(simplejson.dumps(contact_data)) 

class WordCountPipeline(base_handler.PipelineBase):
  """A pipeline to run Word count demo.

  Args:
    blobkey: blobkey to process as string. Should be a zip archive with
      text files inside.
  """

  def run(self, date):
    output = yield mapreduce_pipeline.MapreducePipeline(
        "word_count",
        "main.word_count_map",
        "main.word_count_reduce",
        "mapreduce.input_readers.DatastoreInputReader",
        "mapreduce.output_writers.BlobstoreOutputWriter",
        mapper_params={
            "date": date,
        },
        reducer_params={
            "mime_type": "text/plain",
        },
        shards=16)
    yield StoreOutput("WordCount", output)

class StoreOutput(base_handler.PipelineBase):
  """A pipeline to store the result of the MapReduce job in the database.

  Args:
    mr_type: the type of mapreduce job run (e.g., WordCount, Index)
    encoded_key: the DB key corresponding to the metadata of this job
    output: the blobstore location where the output of the job is stored
  """

  def run(self, mr_type, output):
      logging.info(output) # here I should append the grouped duration in JSON

app engine data processing mapreduce data storage data aggregation query optimization group by contact management

在App Engine中使用MapReduce进行GROUP BY

1 个回答

撰写回答