我（在MongoDB中）如何将多个集合的数据合并到一个集合中？
我可以使用map-reduce吗？如果可以，那么如何使用？
作为新手，我非常需要一些示例。
11个回答
200
MongoDB 3.2 现在允许使用$lookup集合把多个集合中的数据合并成一个。举个实际例子，假设你有关于书籍的数据分散在两个不同的集合中。
第一个集合名为books，包含以下数据：
{
"isbn": "978-3-16-148410-0",
"title": "Some cool book",
"author": "John Doe"
}
{
"isbn": "978-3-16-148999-9",
"title": "Another awesome book",
"author": "Jane Roe"
}
第二个集合名为books_selling_data，其中包含以下数据：
{
"_id": ObjectId("56e31bcf76cdf52e541d9d26"),
"isbn": "978-3-16-148410-0",
"copies_sold": 12500
}
{
"_id": ObjectId("56e31ce076cdf52e541d9d28"),
"isbn": "978-3-16-148999-9",
"copies_sold": 720050
}
{
"_id": ObjectId("56e31ce076cdf52e541d9d29"),
"isbn": "978-3-16-148999-9",
"copies_sold": 1000
}
要合并这两个集合只需要使用如下的 $lookup 方法：
db.books.aggregate([{
$lookup: {
from: "books_selling_data",
localField: "isbn",
foreignField: "isbn",
as: "copies_sold"
}
}])
进行聚合后，books集合将会变成以下形式：
{
"isbn": "978-3-16-148410-0",
"title": "Some cool book",
"author": "John Doe",
"copies_sold": [
{
"_id": ObjectId("56e31bcf76cdf52e541d9d26"),
"isbn": "978-3-16-148410-0",
"copies_sold": 12500
}
]
}
{
"isbn": "978-3-16-148999-9",
"title": "Another awesome book",
"author": "Jane Roe",
"copies_sold": [
{
"_id": ObjectId("56e31ce076cdf52e541d9d28"),
"isbn": "978-3-16-148999-9",
"copies_sold": 720050
},
{
"_id": ObjectId("56e31ce076cdf52e541d9d28"),
"isbn": "978-3-16-148999-9",
"copies_sold": 1000
}
]
}
需要注意以下几点：
- "from"集合（在此示例中为
books_selling_data）不能进行分片。 - "as"字段将是一个数组，就像上面的例子一样。
- $lookup阶段上的"localField"和"foreignField"选项，如果它们在各自的集合中不存在，则在匹配目的上将被视为null（$lookup文档中有关于这个完美的例子）。
因此，总之，如果您想要 consoidate 这两个集合，并且在这种情况下拥有一个名为 copies_sold 的平坦字段来统计总销售量，那么您将不得不做更多的工作，可能需要使用一个中介集合，然后使用$out到最终集合。
- Bruno Krebs
4
你好，能否告诉我如何优化管理像这样的数据：用户、文件.文件和文件.块是三个集合，我想在响应中获取特定用户及其所有相关文件，这可能吗？{
"name": "蝙蝠侠",
"email": "bt@gmail.com",
"files": [
{file1},
{file2},
{file3},
// ...等等
]
} - mfaisalhyder
以上解决方案的官方文档示例可以在此处找到：https://docs.mongodb.com/manual/reference/operator/aggregation/lookup/ - Jakub Czaplicki
4实际上，我的回答已经包含三个官方文档链接了。不过还是谢谢你的贡献。@JakubCzaplicki - Bruno Krebs
2我可能正在经历一次全脑失灵（最有可能的情况），但在
$lookup 中，不是应该要求“localField”和“foreignField”都等于“isbn”吗？而不是“_id”和“isbn”？ - Dev01153
尽管您无法实时执行此操作，但是可以使用 MongoDB 1.8+ 的 map/reduce 中的“reduce”选项多次运行 map-reduce 来合并数据（请参见http://www.mongodb.org/display/DOCS/MapReduce#MapReduce-Outputoptions）。您需要在两个集合中都有一些关键字可用作 _id。
例如，假设您有一个用户集合和一个评论集合，您希望有一个新的集合，其中包含每个评论的一些用户人口统计信息。
假设用户集合具有以下字段：
_id firstName lastName country gender age 然后，评论集合具有以下字段：
_id userId comment created
您将执行以下 map/reduce：
在这一点上，您将拥有一个名为
这是一个相对简单的例子。您可以重复此过程，使用更多的集合来构建减少的集合。您还可以在过程中对数据进行汇总和聚合。随着聚合和保留现有字段的逻辑变得更加复杂，可能会定义多个reduce函数。
您还会注意到，现在每个用户都有一个文档，其中包含该用户的所有评论数组。如果我们合并具有一对一关系而不是一对多关系的数据，它将是平面的，您可以使用像这样的减少函数：
如果您想将
例如，假设您有一个用户集合和一个评论集合，您希望有一个新的集合，其中包含每个评论的一些用户人口统计信息。
假设用户集合具有以下字段：
_id firstName lastName country gender age 然后，评论集合具有以下字段：
_id userId comment created
您将执行以下 map/reduce：
var mapUsers, mapComments, reduce;
db.users_comments.remove();
// setup sample data - wouldn't actually use this in production
db.users.remove();
db.comments.remove();
db.users.save({firstName:"Rich",lastName:"S",gender:"M",country:"CA",age:"18"});
db.users.save({firstName:"Rob",lastName:"M",gender:"M",country:"US",age:"25"});
db.users.save({firstName:"Sarah",lastName:"T",gender:"F",country:"US",age:"13"});
var users = db.users.find();
db.comments.save({userId: users[0]._id, "comment": "Hey, what's up?", created: new ISODate()});
db.comments.save({userId: users[1]._id, "comment": "Not much", created: new ISODate()});
db.comments.save({userId: users[0]._id, "comment": "Cool", created: new ISODate()});
// end sample data setup
mapUsers = function() {
var values = {
country: this.country,
gender: this.gender,
age: this.age
};
emit(this._id, values);
};
mapComments = function() {
var values = {
commentId: this._id,
comment: this.comment,
created: this.created
};
emit(this.userId, values);
};
reduce = function(k, values) {
var result = {}, commentFields = {
"commentId": '',
"comment": '',
"created": ''
};
values.forEach(function(value) {
var field;
if ("comment" in value) {
if (!("comments" in result)) {
result.comments = [];
}
result.comments.push(value);
} else if ("comments" in value) {
if (!("comments" in result)) {
result.comments = [];
}
result.comments.push.apply(result.comments, value.comments);
}
for (field in value) {
if (value.hasOwnProperty(field) && !(field in commentFields)) {
result[field] = value[field];
}
}
});
return result;
};
db.users.mapReduce(mapUsers, reduce, {"out": {"reduce": "users_comments"}});
db.comments.mapReduce(mapComments, reduce, {"out": {"reduce": "users_comments"}});
db.users_comments.find().pretty(); // see the resulting collection
在这一点上，您将拥有一个名为
users_comments的新集合，其中包含合并的数据，您现在可以使用它。这些减少的集合都有一个_id，它是您在映射函数中发出的键，然后所有值都是value键内的子对象 - 这些缩小文档的顶层没有值。这是一个相对简单的例子。您可以重复此过程，使用更多的集合来构建减少的集合。您还可以在过程中对数据进行汇总和聚合。随着聚合和保留现有字段的逻辑变得更加复杂，可能会定义多个reduce函数。
您还会注意到，现在每个用户都有一个文档，其中包含该用户的所有评论数组。如果我们合并具有一对一关系而不是一对多关系的数据，它将是平面的，您可以使用像这样的减少函数：
reduce = function(k, values) {
var result = {};
values.forEach(function(value) {
var field;
for (field in value) {
if (value.hasOwnProperty(field)) {
result[field] = value[field];
}
}
});
return result;
};
如果您想将
users_comments集合展平为每个评论一个文档，还需运行以下命令：var map, reduce;
map = function() {
var debug = function(value) {
var field;
for (field in value) {
print(field + ": " + value[field]);
}
};
debug(this);
var that = this;
if ("comments" in this.value) {
this.value.comments.forEach(function(value) {
emit(value.commentId, {
userId: that._id,
country: that.value.country,
age: that.value.age,
comment: value.comment,
created: value.created,
});
});
}
};
reduce = function(k, values) {
var result = {};
values.forEach(function(value) {
var field;
for (field in value) {
if (value.hasOwnProperty(field)) {
result[field] = value[field];
}
}
});
return result;
};
db.users_comments.mapReduce(map, reduce, {"out": "comments_with_demographics"});
这种技术绝对不能在运行时执行。它适用于定期更新合并数据的cron job或类似的任务。你可能需要在新集合上运行ensureIndex，以确保针对它执行的查询运行速度快（请记住，你的数据仍然在value键内，因此如果你要按评论created时间对comments_with_demographics进行索引，它应该是db.comments_with_demographics.ensureIndex({"value.created": 1});）。
- rmarscher
5
1我在生产软件中可能永远不会这样做，但这仍然是一种非常酷的技术。 - Dave Griffith
3谢谢，Dave。我使用这种技术生成出口和报告表格，在高负载的生产网站上已经使用了3个月，没有问题。这里有另一篇文章，描述了类似的技术应用：http://tebros.com/2011/07/using-mongodb-mapreduce-to-join-2-collections/。 - rmarscher
1谢谢@rmarscher，你提供的额外细节真的帮助我更好地理解了一切。 - benstr
5我应该更新这个答案，用聚合管道和新的$lookup操作举一个例子。在我能组织好一篇适当的文章之前，在这里提到它。https://docs.mongodb.org/manual/reference/operator/aggregation/lookup/ - rmarscher
1对于那些想要快速了解这个程序的人，以下是第一段代码块后
users_comments集合中的内容。https://gist.github.com/nolanamy/83d7fb6a9bf92482a1c4311ad9c78835 - Nolan Amy46
使用聚合和查找，可以在MongoDB中以“SQL UNION”的方式进行联合查询。以下是我在MongoDB 4.0上测试过的示例：
// Create employees data for testing the union.
db.getCollection('employees').insert({ name: "John", type: "employee", department: "sales" });
db.getCollection('employees').insert({ name: "Martha", type: "employee", department: "accounting" });
db.getCollection('employees').insert({ name: "Amy", type: "employee", department: "warehouse" });
db.getCollection('employees').insert({ name: "Mike", type: "employee", department: "warehouse" });
// Create freelancers data for testing the union.
db.getCollection('freelancers').insert({ name: "Stephany", type: "freelancer", department: "accounting" });
db.getCollection('freelancers').insert({ name: "Martin", type: "freelancer", department: "sales" });
db.getCollection('freelancers').insert({ name: "Doug", type: "freelancer", department: "warehouse" });
db.getCollection('freelancers').insert({ name: "Brenda", type: "freelancer", department: "sales" });
// Here we do a union of the employees and freelancers using a single aggregation query.
db.getCollection('freelancers').aggregate( // 1. Use any collection containing at least one document.
[
{ $limit: 1 }, // 2. Keep only one document of the collection.
{ $project: { _id: '$$REMOVE' } }, // 3. Remove everything from the document.
// 4. Lookup collections to union together.
{ $lookup: { from: 'employees', pipeline: [{ $match: { department: 'sales' } }], as: 'employees' } },
{ $lookup: { from: 'freelancers', pipeline: [{ $match: { department: 'sales' } }], as: 'freelancers' } },
// 5. Union the collections together with a projection.
{ $project: { union: { $concatArrays: ["$employees", "$freelancers"] } } },
// 6. Unwind and replace root so you end up with a result set.
{ $unwind: '$union' },
{ $replaceRoot: { newRoot: '$union' } }
]);
以下是它如何工作的解释：
实例化一个
聚合集合，该集合可以由数据库中具有至少一个文档的任何集合构成。如果无法保证您的数据库的任何集合都不会为空，则可以通过在数据库中创建某种“虚拟”集合来解决此问题，其中包含一个单独的空文档，专门用于执行联合查询。将管道的第一个阶段设置为
{ $limit: 1 }。这将剥离集合中除第一个文档以外的所有文档。使用
$project阶段剥离剩余文档的所有字段：{ $project: { _id: '$$REMOVE' } }你的聚合现在只包含一个空文档。现在是时候为想要合并在一起的每个集合添加查找了。您可以使用
pipeline字段进行特定的过滤，或将localField和foreignField设为 null 以匹配整个集合。{ Collection1: [...], Collection2: [...], Collection3: [...] }接着使用
$project阶段和$concatArrays聚合运算符，将它们合并成一个单一的数组：{ "$project" : { "Union" : { $concatArrays: ["$Collection1", "$Collection2", "$Collection3"] } } }现在您拥有一个聚合，其中包含一个文档，其中包含一个包含您收集并集的数组。剩下要做的就是添加一个
$unwind和一个$replaceRoot阶段，将数组拆分为单独的文档：现在你已经拥有了一个包含你想要合并的集合的结果集。你可以添加更多的阶段来进一步过滤、排序、应用skip()和limit()，几乎任何你想要的操作。
{ $lookup: { from: 'collectionToUnion1', pipeline: [...], as: 'Collection1' } },
{ $lookup: { from: 'collectionToUnion2', pipeline: [...], as: 'Collection2' } },
{ $lookup: { from: 'collectionToUnion3', pipeline: [...], as: 'Collection3' } }
{ $unwind: "$Union" },
{ $replaceRoot: { newRoot: "$Union" } }
- sboisse
11
1感谢您对每个步骤进行如此详细的解释。 - Kevin Südmersen
1@sboisse 这个查询在大型集合上的性能如何？ - Ankita
3我的个人经验表明，这种方法在性能方面迄今为止非常令人满意。但是如果你需要像 SQL UNION 那样聚合数据，我看不出还有其他选择。如果你在使用此方法时遇到性能问题，建议优化查询管道中的查找操作，并为查找的集合添加适当的索引。在管道的初始阶段过滤掉的数据越多，效果就越好。在第一步中，我也建议尝试选择一个小的集合，例如只包含一个文档的集合，以便这一步尽可能快速。 - sboisse
@ankita - 很抱歉我没有在Java上工作过。你应该将其作为一个单独的Stack Overflow问题，这是不相关的。 - sboisse
你的方法独一无二，我已经找了很多小时了，没有在其他地方看到过，你是怎么想出来的？ - Ali80
显示剩余6条评论
21
从 Mongo 4.4 开始，我们可以通过将新的 $unionWith 聚合阶段与 $group 的新的 $accumulator 运算符相结合，在聚合管道中实现该联接：
// > db.users.find()
// [{ user: 1, name: "x" }, { user: 2, name: "y" }]
// > db.books.find()
// [{ user: 1, book: "a" }, { user: 1, book: "b" }, { user: 2, book: "c" }]
// > db.movies.find()
// [{ user: 1, movie: "g" }, { user: 2, movie: "h" }, { user: 2, movie: "i" }]
db.users.aggregate([
{ $unionWith: "books" },
{ $unionWith: "movies" },
{ $group: {
_id: "$user",
user: {
$accumulator: {
accumulateArgs: ["$name", "$book", "$movie"],
init: function() { return { books: [], movies: [] } },
accumulate: function(user, name, book, movie) {
if (name) user.name = name;
if (book) user.books.push(book);
if (movie) user.movies.push(movie);
return user;
},
merge: function(userV1, userV2) {
if (userV2.name) userV1.name = userV2.name;
userV1.books.concat(userV2.books);
userV1.movies.concat(userV2.movies);
return userV1;
},
lang: "js"
}
}
}}
])
// { _id: 1, user: { books: ["a", "b"], movies: ["g"], name: "x" } }
// { _id: 2, user: { books: ["c"], movies: ["h", "i"], name: "y" } }
$unionWith用于将给定集合中的记录与聚合管道中已有的文档组合。在两个union阶段之后，我们就可以在管道中找到所有用户、书籍和电影的记录。然后我们使用
$group按$user分组记录，并使用$accumulator操作符累积项，允许在对文档进行分组时进行自定义累积：- 我们要累积的字段是通过
accumulateArgs定义的。 init定义了我们将在分组元素时累积的状态。accumulate函数允许执行一个自定义操作来构建累积状态，例如，如果正在分组的项目已经定义了book字段，则我们会更新状态的books部分。merge用于合并两个内部状态。仅当聚合运行在分片集群上或操作超过内存限制时才使用。
- 我们要累积的字段是通过
- Xavier Guihot
19
$lookup的非常基础的示例。
db.getCollection('users').aggregate([
{
$lookup: {
from: "userinfo",
localField: "userId",
foreignField: "userId",
as: "userInfoData"
}
},
{
$lookup: {
from: "userrole",
localField: "userId",
foreignField: "userId",
as: "userRoleData"
}
},
{ $unwind: { path: "$userInfoData", preserveNullAndEmptyArrays: true }},
{ $unwind: { path: "$userRoleData", preserveNullAndEmptyArrays: true }}
])
这里使用了
{ $unwind: { path: "$userInfoData", preserveNullAndEmptyArrays: true }},
{ $unwind: { path: "$userRoleData", preserveNullAndEmptyArrays: true }}
不是
{ $unwind:"$userRoleData"}
{ $unwind:"$userRoleData"}
由于{ $unwind:"$userRoleData"}，如果使用$lookup没有找到匹配记录，将返回空或0个结果。
- Anish Agarwal
14
在聚合操作中使用多个$lookup查询不同的集合。
查询：
db.getCollection('servicelocations').aggregate([
{
$match: {
serviceLocationId: {
$in: ["36728"]
}
}
},
{
$lookup: {
from: "orders",
localField: "serviceLocationId",
foreignField: "serviceLocationId",
as: "orders"
}
},
{
$lookup: {
from: "timewindowtypes",
localField: "timeWindow.timeWindowTypeId",
foreignField: "timeWindowTypeId",
as: "timeWindow"
}
},
{
$lookup: {
from: "servicetimetypes",
localField: "serviceTimeTypeId",
foreignField: "serviceTimeTypeId",
as: "serviceTime"
}
},
{
$unwind: "$orders"
},
{
$unwind: "$serviceTime"
},
{
$limit: 14
}
])
结果：
{
"_id" : ObjectId("59c3ac4bb7799c90ebb3279b"),
"serviceLocationId" : "36728",
"regionId" : 1.0,
"zoneId" : "DXBZONE1",
"description" : "AL HALLAB REST EMIRATES MALL",
"locationPriority" : 1.0,
"accountTypeId" : 1.0,
"locationType" : "SERVICELOCATION",
"location" : {
"makani" : "",
"lat" : 25.119035,
"lng" : 55.198694
},
"deliveryDays" : "MTWRFSU",
"timeWindow" : [
{
"_id" : ObjectId("59c3b0a3b7799c90ebb32cde"),
"timeWindowTypeId" : "1",
"Description" : "MORNING",
"timeWindow" : {
"openTime" : "06:00",
"closeTime" : "08:00"
},
"accountId" : 1.0
},
{
"_id" : ObjectId("59c3b0a3b7799c90ebb32cdf"),
"timeWindowTypeId" : "1",
"Description" : "MORNING",
"timeWindow" : {
"openTime" : "09:00",
"closeTime" : "10:00"
},
"accountId" : 1.0
},
{
"_id" : ObjectId("59c3b0a3b7799c90ebb32ce0"),
"timeWindowTypeId" : "1",
"Description" : "MORNING",
"timeWindow" : {
"openTime" : "10:30",
"closeTime" : "11:30"
},
"accountId" : 1.0
}
],
"address1" : "",
"address2" : "",
"phone" : "",
"city" : "",
"county" : "",
"state" : "",
"country" : "",
"zipcode" : "",
"imageUrl" : "",
"contact" : {
"name" : "",
"email" : ""
},
"status" : "ACTIVE",
"createdBy" : "",
"updatedBy" : "",
"updateDate" : "",
"accountId" : 1.0,
"serviceTimeTypeId" : "1",
"orders" : [
{
"_id" : ObjectId("59c3b291f251c77f15790f92"),
"orderId" : "AQ18O1704264",
"serviceLocationId" : "36728",
"orderNo" : "AQ18O1704264",
"orderDate" : "18-Sep-17",
"description" : "AQ18O1704264",
"serviceType" : "Delivery",
"orderSource" : "Import",
"takenBy" : "KARIM",
"plannedDeliveryDate" : ISODate("2017-08-26T00:00:00.000Z"),
"plannedDeliveryTime" : "",
"actualDeliveryDate" : "",
"actualDeliveryTime" : "",
"deliveredBy" : "",
"size1" : 296.0,
"size2" : 3573.355,
"size3" : 240.811,
"jobPriority" : 1.0,
"cancelReason" : "",
"cancelDate" : "",
"cancelBy" : "",
"reasonCode" : "",
"reasonText" : "",
"status" : "",
"lineItems" : [
{
"ItemId" : "BNWB020",
"size1" : 15.0,
"size2" : 78.6,
"size3" : 6.0
},
{
"ItemId" : "BNWB021",
"size1" : 20.0,
"size2" : 252.0,
"size3" : 11.538
},
{
"ItemId" : "BNWB023",
"size1" : 15.0,
"size2" : 285.0,
"size3" : 16.071
},
{
"ItemId" : "CPMW112",
"size1" : 3.0,
"size2" : 25.38,
"size3" : 1.731
},
{
"ItemId" : "MMGW001",
"size1" : 25.0,
"size2" : 464.375,
"size3" : 46.875
},
{
"ItemId" : "MMNB218",
"size1" : 50.0,
"size2" : 920.0,
"size3" : 60.0
},
{
"ItemId" : "MMNB219",
"size1" : 50.0,
"size2" : 630.0,
"size3" : 40.0
},
{
"ItemId" : "MMNB220",
"size1" : 50.0,
"size2" : 416.0,
"size3" : 28.846
},
{
"ItemId" : "MMNB270",
"size1" : 50.0,
"size2" : 262.0,
"size3" : 20.0
},
{
"ItemId" : "MMNB302",
"size1" : 15.0,
"size2" : 195.0,
"size3" : 6.0
},
{
"ItemId" : "MMNB373",
"size1" : 3.0,
"size2" : 45.0,
"size3" : 3.75
}
],
"accountId" : 1.0
},
{
"_id" : ObjectId("59c3b291f251c77f15790f9d"),
"orderId" : "AQ137O1701240",
"serviceLocationId" : "36728",
"orderNo" : "AQ137O1701240",
"orderDate" : "18-Sep-17",
"description" : "AQ137O1701240",
"serviceType" : "Delivery",
"orderSource" : "Import",
"takenBy" : "KARIM",
"plannedDeliveryDate" : ISODate("2017-08-26T00:00:00.000Z"),
"plannedDeliveryTime" : "",
"actualDeliveryDate" : "",
"actualDeliveryTime" : "",
"deliveredBy" : "",
"size1" : 28.0,
"size2" : 520.11,
"size3" : 52.5,
"jobPriority" : 1.0,
"cancelReason" : "",
"cancelDate" : "",
"cancelBy" : "",
"reasonCode" : "",
"reasonText" : "",
"status" : "",
"lineItems" : [
{
"ItemId" : "MMGW001",
"size1" : 25.0,
"size2" : 464.38,
"size3" : 46.875
},
{
"ItemId" : "MMGW001-F1",
"size1" : 3.0,
"size2" : 55.73,
"size3" : 5.625
}
],
"accountId" : 1.0
},
{
"_id" : ObjectId("59c3b291f251c77f15790fd8"),
"orderId" : "AQ110O1705036",
"serviceLocationId" : "36728",
"orderNo" : "AQ110O1705036",
"orderDate" : "18-Sep-17",
"description" : "AQ110O1705036",
"serviceType" : "Delivery",
"orderSource" : "Import",
"takenBy" : "KARIM",
"plannedDeliveryDate" : ISODate("2017-08-26T00:00:00.000Z"),
"plannedDeliveryTime" : "",
"actualDeliveryDate" : "",
"actualDeliveryTime" : "",
"deliveredBy" : "",
"size1" : 60.0,
"size2" : 1046.0,
"size3" : 68.0,
"jobPriority" : 1.0,
"cancelReason" : "",
"cancelDate" : "",
"cancelBy" : "",
"reasonCode" : "",
"reasonText" : "",
"status" : "",
"lineItems" : [
{
"ItemId" : "MMNB218",
"size1" : 50.0,
"size2" : 920.0,
"size3" : 60.0
},
{
"ItemId" : "MMNB219",
"size1" : 10.0,
"size2" : 126.0,
"size3" : 8.0
}
],
"accountId" : 1.0
}
],
"serviceTime" : {
"_id" : ObjectId("59c3b07cb7799c90ebb32cdc"),
"serviceTimeTypeId" : "1",
"serviceTimeType" : "nohelper",
"description" : "",
"fixedTime" : 30.0,
"variableTime" : 0.0,
"accountId" : 1.0
}
}
- KARTHIKEYAN.A
14
如果没有批量插入到 MongoDB 中，我们会循环处理
small_collection 中的所有对象，并逐个将它们插入到 big_collection 中。db.small_collection.find().forEach(function(obj){
db.big_collection.insert(obj)
});
- Hieu Le
2
1db.collection.insert([{},{},{}])Insert方法接受数组。 - augurone
2这对于小集合来说很好用，但不要忘记迁移索引 :) - Sebastien Lorber
1
Mongorestore具有将内容附加到数据库中已有内容之上的功能，因此可以使用此行为来合并两个集合：
- mongodump collection1
- collection2.rename(collection1)
- mongorestore
还没有尝试过，但可能比map/reduce方法更快。
- shauli
0
可以的：使用我今天编写的这个实用函数：
function shangMergeCol() {
tcol= db.getCollection(arguments[0]);
for (var i=1; i<arguments.length; i++){
scol= db.getCollection(arguments[i]);
scol.find().forEach(
function (d) {
tcol.insert(d);
}
)
}
}
您可以向此函数传递任意数量的集合，第一个集合将成为目标集合。所有其他集合都是要转移到目标集合的源。
- Shangab
-1
代码片段。感谢Stack Overflow上的多篇帖子，包括这篇。
db.cust.drop();
db.zip.drop();
db.cust.insert({cust_id:1, zip_id: 101});
db.cust.insert({cust_id:2, zip_id: 101});
db.cust.insert({cust_id:3, zip_id: 101});
db.cust.insert({cust_id:4, zip_id: 102});
db.cust.insert({cust_id:5, zip_id: 102});
db.zip.insert({zip_id:101, zip_cd:'AAA'});
db.zip.insert({zip_id:102, zip_cd:'BBB'});
db.zip.insert({zip_id:103, zip_cd:'CCC'});
mapCust = function() {
var values = {
cust_id: this.cust_id
};
emit(this.zip_id, values);
};
mapZip = function() {
var values = {
zip_cd: this.zip_cd
};
emit(this.zip_id, values);
};
reduceCustZip = function(k, values) {
var result = {};
values.forEach(function(value) {
var field;
if ("cust_id" in value) {
if (!("cust_ids" in result)) {
result.cust_ids = [];
}
result.cust_ids.push(value);
} else {
for (field in value) {
if (value.hasOwnProperty(field) ) {
result[field] = value[field];
}
};
}
});
return result;
};
db.cust_zip.drop();
db.cust.mapReduce(mapCust, reduceCustZip, {"out": {"reduce": "cust_zip"}});
db.zip.mapReduce(mapZip, reduceCustZip, {"out": {"reduce": "cust_zip"}});
db.cust_zip.find();
mapCZ = function() {
var that = this;
if ("cust_ids" in this.value) {
this.value.cust_ids.forEach(function(value) {
emit(value.cust_id, {
zip_id: that._id,
zip_cd: that.value.zip_cd
});
});
}
};
reduceCZ = function(k, values) {
var result = {};
values.forEach(function(value) {
var field;
for (field in value) {
if (value.hasOwnProperty(field)) {
result[field] = value[field];
}
}
});
return result;
};
db.cust_zip_joined.drop();
db.cust_zip.mapReduce(mapCZ, reduceCZ, {"out": "cust_zip_joined"});
db.cust_zip_joined.find().pretty();
var flattenMRCollection=function(dbName,collectionName) {
var collection=db.getSiblingDB(dbName)[collectionName];
var i=0;
var bulk=collection.initializeUnorderedBulkOp();
collection.find({ value: { $exists: true } }).addOption(16).forEach(function(result) {
print((++i));
//collection.update({_id: result._id},result.value);
bulk.find({_id: result._id}).replaceOne(result.value);
if(i%1000==0)
{
print("Executing bulk...");
bulk.execute();
bulk=collection.initializeUnorderedBulkOp();
}
});
bulk.execute();
};
flattenMRCollection("mydb","cust_zip_joined");
db.cust_zip_joined.find().pretty();
- Vipul Mehta
网页内容由stack overflow 提供, 点击上面的可以查看英文原文，
原文链接
原文链接
db.collection1.find().forEach(function(doc){db.collection2.save(doc)});即可。如果你不使用mongo shell，请指明你使用的驱动程序（如java、php等）。 - proximus