javascript - mongo/mongoid MapReduce 批量插入文档

Question

我正在创建我的批处理并使用下面指定的命令将其插入集合中

batch = []
time = 1.day.ago

(1..2000).each{ |i| a = {:name => 'invbatch2k'+i.to_s, :user_id =>  BSON::ObjectId.from_string('533956cd4d616323cf000000'), :out_id => 'out', :created_at => time, :updated_at => time, :random => '0.5' }; batch.push a; }

Invitation.collection.insert batch

如上所述，每条邀请记录的 user_id 字段值均设置为 '533956cd4d616323cf000000'

使用 created_at: 1.day.ago 插入我的批处理后，我得到:

2.1.1 :102 > Invitation.lte(created_at: 1.week.ago).count
 => 48
2.1.1 :103 > Invitation.lte(created_at: Date.today).count
 => 2048

还有:

2.1.1 :104 > Invitation.lte(created_at: 1.week.ago).where(user_id: '533956cd4d616323cf000000').count
 => 14
2.1.1 :105 > Invitation.where(user_id: '533956cd4d616323cf000000').count
 => 2014

此外，我还有一个 map 缩减功能，可以对每个唯一用户发送的邀请进行计数(包括总数和发送到唯一 out_id 的邀请)

class Invitation

  [...]

  def self.get_user_invites_count
    map = %q{
      function() {
        var user_id = this.user_id;
        emit(user_id, {user_id : this.user_id, out_id: this.out_id, count: 1, countUnique: 1})
      }
    }
    reduce = %q{
      function(key, values) {
        var result = {
          user_id: key,
          count: 0,
          countUnique : 0
        };
        var values_arr = [];
        values.forEach(function(value) {
          values_arr.push(value.out_id);
          result.count += 1
        });
        var unique = values_arr.filter(function(item, i, ar){ return ar.indexOf(item) === i; });
        result.countUnique = unique.length;
        return result;
      }
    }
    map_reduce(map,reduce).out(inline: true).to_a.map{|d| d['value']} rescue []
  end
end

问题是:

Invitation.lte(created_at: Date.today.end_of_day).get_user_invites_count

返回

[{"user_id"=>BSON::ObjectId('533956cd4d616323cf000000'), "count"=>49.0, "countUnique"=>2.0} ...]

而不是“count”=> 2014，“countUnique”=> 6.0同时:

Invitation.lte(created_at: 1.week.ago).get_user_invites_count 返回:

[{"user_id"=>BSON::ObjectId('533956cd4d616323cf000000'), "count"=>14.0, "countUnique"=>6.0} ...]

查询提供的数据在插入批处理之前是准确的。

我无法理解这里发生的事情。我错过了什么吗？

Answer 1

您似乎在 documentation 中错过的部分问题似乎出在这里:

MongoDB can invoke the reduce function more than once for the same key. In this case, the previous output from the reduce function for that key will become one of the input values to the next reduce function invocation for that key.

以及稍后:

the type of the return object must be identical to the type of the value emitted by the map function to ensure that the following operations is true:

所以你看到的是你的reduce函数返回的签名与它从映射器接收到的输入不同。这很重要，因为 reducer 可能无法在一次传递中获得给定键的所有值。相反，它会获取其中的一些值，“减少”结果，并且减少的输出可以在进一步通过归约函数时与键的其他值(可能也减少)组合。

由于您的字段不匹配，后续的reduce过程看不到这些值，并且不会计入您的总数。因此，您需要对齐值的签名:

  def self.get_user_invites_count
    map = %q{
      function() {
        var user_id = this.user_id;
        emit(user_id, {out_id: this.out_id, count: 1, countUnique: 0})
      }
    }
    reduce = %q{
      function(key, values) {
        var result = {
          out_id: null,
          count: 0,
          countUnique : 0
        };
        var values_arr = [];
        values.forEach(function(value) {
          if (value.out_id != null)
            values_arr.push(value.out_id);
          result.count += value.count;
          result.countUnique += value.countUnique;
        });
        var unique = values_arr.filter(function(item, i, ar){ return ar.indexOf(item) === i; });
        result.countUnique += unique.length;
        return result;
      }
    }
    map_reduce(map,reduce).out(inline: true).to_a.map{|d| d['value']} rescue []
  end

您也不需要在发出或保留的值中使用user_id，因为它已经是mapReduce 的“键”值。其余的更改认为“count”和“countUnique”都可以包含需要考虑的现有值，您只需在每次传递时将该值重置为 0。

当然，如果“输入”已经通过“reduce”传递，那么您不需要过滤“out_id”值的“唯一性”，因为您已经有了计数并且现在已包含在内。因此，任何 null 值都不会添加到要计数的数组中，该数组也会“添加”到总数中，而不是替换它。

所以 reducer 确实被调用了几次。对于 20 个键值，输入可能不会被拆分，这就是为什么输入较少的示例有效的原因。除此之外，相同键值的“组”将被拆分，这就是 MapReduce 针对大数据处理进行优化的方式。由于“减少的”输出将再次发送回 reducer ，因此您需要注意，您正在考虑在上一次传递中已发送到输出的值。