向MongoDB中的现有集合添加具有大量行的新字段

Question

我有一个现有的集合，其中包含近 100 万个文档，现在我想在这个集合中附加一个新的字段数据。 （我正在使用 PyMongo）

例如，我现有的集合db.actions 看起来像：

...
{'_id':12345, 'A': 'apple', 'B': 'milk'}
{'_id':12346, 'A': 'pear', 'B': 'juice'}
...

现在我想将一个新的列字段数据附加到这个现有的集合中：

...
{'_id':12345, 'C': 'beef'}
{'_id':12346, 'C': 'chicken'}
...

这样生成的集合应如下所示：

...
{'_id':12345, 'A': 'apple', 'B': 'milk', 'C': 'beef'}
{'_id':12346, 'A': 'pear', 'B': 'juice', 'C': 'chicken'}
...

我知道我们可以用 update_one 和 for 循环来做到这一点，例如

for doc in values:
        collection.update_one({'_id': doc['_id']},
        {'$set': {k: doc[k] for k in fields}},
        upsert=True
    )

其中values 是一个字典列表，每个字典包含两项，_id 键值对和新字段键值对。 fields 包含我想添加的所有新字段。

但是，问题是我有一百万个文档要更新，任何带有for 循环的东西都太慢了，有没有办法更快地附加这个新字段？类似于insert_many 的东西，除了它附加到现有集合？

================================================

更新1：

这就是我现在所拥有的，

bulk = self.get_collection().initialize_unordered_bulk_op()
for doc in values:
    bulk.find({'_id': doc['_id']}).update_one({'$set': {k: doc[k] for k in fields} })

bulk.execute()

我首先使用insert_many将示例数据框写入数据库，性能： Time spent in insert_many: total: 0.0457min 然后我使用update_one 和bulk 操作将额外的两个字段添加到集合中，我得到： Time spent: for loop: 0.0283min | execute: 0.0713min | total: 0.0996min

更新2：

我在现有集合和新列数据中都添加了一个额外的列，目的是使用左连接来解决这个问题。如果您使用左连接，则可以忽略 _id 字段。

例如，我现有的集合db.actions 看起来像：

...
{'A': 'apple', 'B': 'milk', 'dateTime': '2017-10-12 15:20:00'}
{'A': 'pear', 'B': 'juice', 'dateTime': '2017-12-15 06:10:50'}
{'A': 'orange', 'B': 'pop', 'dateTime': '2017-12-15 16:09:10'}
...

现在我想将一个新的列字段数据附加到这个现有的集合中：

...
{'C': 'beef', 'dateTime': '2017-10-12 09:08:20'}
{'C': 'chicken', 'dateTime': '2017-12-15 22:40:00'}
...

这样生成的集合应如下所示：

...
{'A': 'apple', 'B': 'milk', 'C': 'beef', 'dateTime': '2017-10-12'}
{'A': 'pear', 'B': 'juice', 'C': 'chicken', 'dateTime': '2017-12-15'}
{'A': 'orange', 'B': 'pop', 'C': 'chicken', 'dateTime': '2017-12-15'}
...

Answer 1

如果每个文档的更新确实是独一无二的，那么没有什么比 bulk write API 更快的了。 MongoDB 和驱动程序都无法猜测您要更新的内容，因此您需要遍历更新定义，然后批量更改，这在此处进行了详细描述：

Bulk update in Pymongo using multiple ObjectId

“无序”批量写入可能会稍微快一些（尽管在我的测试中不是这样），但我仍然主要出于错误处理原因投票支持有序方法。

但是，如果您可以将更改分组为特定的重复模式，那么您当然最好定义一堆更新查询（实际上是字典中每个唯一值的一次更新），然后发出每个针对多个文档的查询.在这一点上，我的 Python 太差了，无法为您编写整个代码，但这里有一个伪代码示例，说明我的意思：

假设您有以下更新字典：

{
    key: "doc1",
    value:
    [
        { "field1", "value1" },
        { "field2", "value2" },
    ]
}, {
    key: "doc2",
    value:
    [
        // same fields again as for "doc1"
        { "field1", "value1" },
        { "field2", "value2" },
    ]
}, {
    key: "doc3",
    value:
    [
        { "someotherfield", "someothervalue" },
    ]
}

然后，您无需分别更新三个文档，而是发送一个更新来更新前两个文档（因为它们需要相同的更改），然后发送一个更新来更新“doc3”。您对更新模式结构的了解越多，即使通过对字段子集的更新进行分组，您也可以越多地对其进行优化，但这在某些时候可能会变得有点复杂......

更新：

根据您的以下要求，让我们试一试。

fields = ['C']
values = [
    {'_id': 'doc1a', 'C': 'v1'},
    {'_id': 'doc1b', 'C': 'v1'},
    {'_id': 'doc2a', 'C': 'v2'},
    {'_id': 'doc2b', 'C': 'v2'}
]

print 'before transformation:'
for doc in values:
    print('_id ' + doc['_id'])
    for k in fields:
        print(doc[k])

transposed_values = {}
for doc in values:
    transposed_values[doc['C']] = transposed_values.get(doc['C'], [])
    transposed_values[doc['C']].append(doc['_id'])

print 'after transformation:'
for k, v in transposed_values.iteritems():
    print k, v

for k, v in transposed_values.iteritems():
    collection.update_many({'_id': { '$in': v}}, {'$set': {'C': k}})

Answer 2

由于您的联接集合的文档较少，您可以将 dateTime 转换为日期

db.new.find().forEach(function(d){
    d.date = d.dateTime.substring(0,10);
    db.new.update({_id : d._id}, d);
})

并根据日期（dateTime 的子字符串）和 _id 进行多字段查找，

到一个新的集合（增强）

db.old.aggregate(
    [
        {$lookup: {
                from : "new",
                let : {id : "$_id", date : {$substr : ["$dateTime", 0, 10]}},
                pipeline : [
                    {$match : {
                        $expr : {
                            $and : [
                                {$eq : ["$$id", "$_id"]},
                                {$eq : ["$$date", "$date"]}
                            ]
                        }
                    }},
                    {$project : {_id : 0, C : "$C"}}
                ],
                as : "newFields"
            }
        },
        {$project : {
            _id : 1,
            A : 1,
            B : 1,
            C : {$arrayElemAt : ["$newFields.C", 0]},
            date : {$substr : ["$dateTime", 0, 10]}
        }},
        {$out : "enhanced"}
    ]
).pretty()

结果

> db.enhanced.find()
{ "_id" : 12345, "A" : "apple", "B" : "milk", "C" : "beef", "date" : "2017-10-12" }
{ "_id" : 12346, "A" : "pear", "B" : "juice", "C" : "chicken", "date" : "2017-12-15" }
{ "_id" : 12347, "A" : "orange", "B" : "pop", "date" : "2017-12-15" }
>