Пошаговое руководство. Карта MongoDB / Сокращение рабочих мест

Map / Reduce — отличный способ выполнять агрегацию и операции ETL- типа с MongoDB. К сожалению, это может быть немного сложно, когда вы только начинаете. Кайл  сделал отличный пост,  объясняющий основы, и есть пара примеров в Кулинарной книге MongoDB  — вероятно, лучше всего сначала прочитать их, если вы совершенно новичок в map / lower или MongoDB. Во всяком случае, для этого поста я подумал, что было бы полезно просмотреть реальный пример работы с картой / сокращением, которую мы используем в Fiesta .

Эта проблема

Fiesta поддерживает настраиваемые домены (служба с белыми метками) и API для создания списков рассылки   — обе эти функции находятся в закрытой бета-версии ( дайте нам знать,  если вы хотите попробовать их).

Недавно другой стартап начал использовать API для создания списков для одного из своих проектов. Сразу после того, как их служба начала создавать списки, пришел запрос на предоставление аналитических данных о том, как люди используют их списки. Это было то, что мы планировали добавить, но пока не получили. Во всяком случае, я намеревался объединить некоторые аналитические данные для пользовательских доменов.

Данные

По умолчанию Fiesta никогда не сохраняет содержимое сообщений; однако мы храним некоторые метаданные сообщения. Вот пример документа MongoDB, который сохраняется для каждого входящего сообщения:

{
  "_id": oid,
  "from": from_address,
  "group_name": name,
  "domain": domain,
  "group_id": group_id,
  "message_id": message_id,
  "parent_id": parent_id,
  "thread_id": thread_id
}

Поскольку _id поле является ObjectId , он содержит встроенную метку времени. Между отметкой времени и полем домена у нас есть все данные, которые нам нужны, чтобы предоставить некоторую аналитику о том, сколько сообщений было отправлено в определенный пользовательский домен за определенный период времени. Давайте перейдем к карте / уменьшить, что мы используем для вычисления этой аналитики.

Карта / Уменьшить

Давайте начнем с рассмотрения нашей функции map ():

function map () {
  var ts = this._id.getTimestamp();
  var month = ts.getUTCFullYear() + "-" + (ts.getUTCMonth() + 1);
  var stats = {};
  stats[month + "-" + ts.getUTCDate()] = 1;
  emit(this.domain, stats);
}

Функция map () будет вызываться один раз для каждого документа в коллекции «messages». Наша цель — просто сгруппировать каждое сообщение по домену и дате. Мы генерируем () нашу статистику, используя домен сообщения в качестве ключа. Это означает, что после шага Reduce () мы получим один результат на домен. Значение, которое мы излучаем, — это просто дата отправки сообщения, примерно так:

{
  "2011-10-2": 1
}

После завершения шага map () у нас будет куча документов с метками времени, связанных с каждым настраиваемым доменом. Следующий шаг — свести их в один документ:

function reduce(key, values) {
  var out = {};
  function merge(a, b) {
    for (var k in b) {
      if (!b.hasOwnProperty(k)) {
        continue;
      }
      a[k] = (a[k] || 0) + b[k];
    }
  }
  for (var i=0; i < values.length; i++) {
    merge(out, values[i]);
  }
  return out;
}

Ключ в этом случае является проблемно — зависимым, как «example.org», и значением является массивом эмитированных значений для этого ключа. Наша цель — просто просмотреть каждое из выданных значений и суммировать их, чтобы мы получили итоговую сумму за каждую дату. Важная вещь, которую следует запомнить, и причина большинства ошибок отображения / уменьшения состоит в том, что массив значений может содержать значения, выданные во время шага map (), или значения, вычисленные из предыдущей итерации redu () . Мы должны быть осторожны, чтобы обрабатывать оба этих случая в нашей функции Reduce (хороший подход — сохранять случаи одинаковыми — убедитесь, что выходные данные шага сокращения «похожи» на выходные данные шага карты).

{
  "_id": "example.com",
  "value": {
    "2011-9-26": 21.0,
    "2011-9-28": 25.0,
    "2011-10-1": 142.0,
    "2011-10-2": 16.0
  }
}

def compute_domain_analytics():
    """
    Compute statistics about the number of messages being sent to a domain.
    """
    map = """function map () {
  var ts = this._id.getTimestamp();
  var month = ts.getUTCFullYear() + "-" + (ts.getUTCMonth() + 1);
  var stats = {};
  stats[month + "-" + ts.getUTCDate()] = 1;
  emit(this.domain, stats);
}"""

    reduce = """function reduce(key, values) {
  var out = {};
  function merge(a, b) {
    for (var k in b) {
      if (!b.hasOwnProperty(k)) {
        continue;
      }
      a[k] = (a[k] || 0) + b[k];
    }
  }
  for (var i=0; i < values.length; i++) {
    merge(out, values[i]);
  }
  return out;
}"""
    db.db.messages.map_reduce(map, reduce, out="domain_analytics")