批量插入文档上的 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 reduce，它计算每个唯一用户发送的邀请（总数和发送到唯一 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 可以为同一个键多次调用 reduce 函数。在这种情况下，该键的 reduce 函数的先前输出将成为该键的下一个 reduce 函数调用的输入值之一。

还有以后：

返回对象的类型必须与map函数发出的值的类型相同，以确保以下操作为真：

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

由于您的字段不匹配，后续的reduce pass 看不到这些值并且不计入您的总数。所以你需要对齐值的签名：

  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。

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

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