Обобщение на TLDR
В съвременните версии на MongoDB можете да форсирате това с $slice точно извън основния резултат от агрегирането. За „големи“ резултати вместо това стартирайте паралелни заявки за всяко групиране ( демонстрационен списък е в края на отговора ) или изчакайте SERVER-9377 да се разреши, което ще позволи „ограничение“ на броя на елементите за $push към масив.
db.books.aggregate([
{ "$group": {
"_id": {
"addr": "$addr",
"book": "$book"
},
"bookCount": { "$sum": 1 }
}},
{ "$group": {
"_id": "$_id.addr",
"books": {
"$push": {
"book": "$_id.book",
"count": "$bookCount"
},
},
"count": { "$sum": "$bookCount" }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 },
{ "$project": {
"books": { "$slice": [ "$books", 2 ] },
"count": 1
}}
])
Преглед на MongoDB 3.6
Все още не се разрешава SERVER-9377, но в тази версия $lookup позволява нова "некорелирана" опция, която приема "pipeline" израз като аргумент вместо "localFields" и "foreignFields" настроики. Това след това позволява "самоприсъединяване" с друг израз на конвейера, в който можем да приложим $limit за да върнете резултатите "top-n".
db.books.aggregate([
{ "$group": {
"_id": "$addr",
"count": { "$sum": 1 }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 },
{ "$lookup": {
"from": "books",
"let": {
"addr": "$_id"
},
"pipeline": [
{ "$match": {
"$expr": { "$eq": [ "$addr", "$$addr"] }
}},
{ "$group": {
"_id": "$book",
"count": { "$sum": 1 }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 }
],
"as": "books"
}}
])
Другото допълнение тук е, разбира се, възможността за интерполиране на променливата чрез $expr използвайки $match за да изберете съвпадащите елементи в „присъединяване“, но общата предпоставка е „тръбопровод в тръбопровод“, където вътрешното съдържание може да се филтрира чрез съвпадения от родителя. Тъй като и двете сами по себе си са "тръбопроводи", можем да $limit всеки резултат поотделно.
Това би било следващата най-добра опция за изпълнение на паралелни заявки и всъщност би било по-добре, ако $match бяха разрешени и можеха да използват индекс при обработката на "под-тръбопровод". Така че, което е, не използва "лимита до $push " както пита цитираният проблем, той всъщност доставя нещо, което би трябвало да работи по-добре.
Оригинално съдържание
Изглежда, че сте попаднали на горния проблем "N". По някакъв начин вашият проблем е доста лесен за решаване, но не и с точното ограничение, което искате:
db.books.aggregate([
{ "$group": {
"_id": {
"addr": "$addr",
"book": "$book"
},
"bookCount": { "$sum": 1 }
}},
{ "$group": {
"_id": "$_id.addr",
"books": {
"$push": {
"book": "$_id.book",
"count": "$bookCount"
},
},
"count": { "$sum": "$bookCount" }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 }
])
Сега това ще ви даде резултат като този:
{
"result" : [
{
"_id" : "address1",
"books" : [
{
"book" : "book4",
"count" : 1
},
{
"book" : "book5",
"count" : 1
},
{
"book" : "book1",
"count" : 3
}
],
"count" : 5
},
{
"_id" : "address2",
"books" : [
{
"book" : "book5",
"count" : 1
},
{
"book" : "book1",
"count" : 2
}
],
"count" : 3
}
],
"ok" : 1
}
Така че това се различава от това, което питате, докато ние получаваме най-добрите резултати за стойностите на адреса, основният избор на „книги“ не е ограничен само до необходимото количество резултати.
Това се оказва много трудно за изпълнение, но може да се направи, въпреки че сложността просто се увеличава с броя на елементите, които трябва да съпоставите. За да е просто, можем да запазим това най-много 2 съвпадения:
db.books.aggregate([
{ "$group": {
"_id": {
"addr": "$addr",
"book": "$book"
},
"bookCount": { "$sum": 1 }
}},
{ "$group": {
"_id": "$_id.addr",
"books": {
"$push": {
"book": "$_id.book",
"count": "$bookCount"
},
},
"count": { "$sum": "$bookCount" }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 },
{ "$unwind": "$books" },
{ "$sort": { "count": 1, "books.count": -1 } },
{ "$group": {
"_id": "$_id",
"books": { "$push": "$books" },
"count": { "$first": "$count" }
}},
{ "$project": {
"_id": {
"_id": "$_id",
"books": "$books",
"count": "$count"
},
"newBooks": "$books"
}},
{ "$unwind": "$newBooks" },
{ "$group": {
"_id": "$_id",
"num1": { "$first": "$newBooks" }
}},
{ "$project": {
"_id": "$_id",
"newBooks": "$_id.books",
"num1": 1
}},
{ "$unwind": "$newBooks" },
{ "$project": {
"_id": "$_id",
"num1": 1,
"newBooks": 1,
"seen": { "$eq": [
"$num1",
"$newBooks"
]}
}},
{ "$match": { "seen": false } },
{ "$group":{
"_id": "$_id._id",
"num1": { "$first": "$num1" },
"num2": { "$first": "$newBooks" },
"count": { "$first": "$_id.count" }
}},
{ "$project": {
"num1": 1,
"num2": 1,
"count": 1,
"type": { "$cond": [ 1, [true,false],0 ] }
}},
{ "$unwind": "$type" },
{ "$project": {
"books": { "$cond": [
"$type",
"$num1",
"$num2"
]},
"count": 1
}},
{ "$group": {
"_id": "$_id",
"count": { "$first": "$count" },
"books": { "$push": "$books" }
}},
{ "$sort": { "count": -1 } }
])
Така че това всъщност ще ви даде първите 2 „книги“ от първите два „адресни“ записа.
Но за моите пари, останете с първия формуляр и след това просто „нарежете“ елементите от масива, които се връщат, за да вземат първите „N“ елементи.
Демонстрационен код
Демонстрационният код е подходящ за използване с текущи LTS версии на NodeJS от версии v8.x и v10.x. Това е най-вече за async/await синтаксис, но в общия поток няма нищо, което да има такова ограничение и да се адаптира с малки промени към обикновени обещания или дори обратно към обикновена реализация на обратно извикване.
index.js
const { MongoClient } = require('mongodb');
const fs = require('mz/fs');
const uri = 'mongodb://localhost:27017';
const log = data => console.log(JSON.stringify(data, undefined, 2));
(async function() {
try {
const client = await MongoClient.connect(uri);
const db = client.db('bookDemo');
const books = db.collection('books');
let { version } = await db.command({ buildInfo: 1 });
version = parseFloat(version.match(new RegExp(/(?:(?!-).)*/))[0]);
// Clear and load books
await books.deleteMany({});
await books.insertMany(
(await fs.readFile('books.json'))
.toString()
.replace(/\n$/,"")
.split("\n")
.map(JSON.parse)
);
if ( version >= 3.6 ) {
// Non-correlated pipeline with limits
let result = await books.aggregate([
{ "$group": {
"_id": "$addr",
"count": { "$sum": 1 }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 },
{ "$lookup": {
"from": "books",
"as": "books",
"let": { "addr": "$_id" },
"pipeline": [
{ "$match": {
"$expr": { "$eq": [ "$addr", "$$addr" ] }
}},
{ "$group": {
"_id": "$book",
"count": { "$sum": 1 },
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 }
]
}}
]).toArray();
log({ result });
}
// Serial result procesing with parallel fetch
// First get top addr items
let topaddr = await books.aggregate([
{ "$group": {
"_id": "$addr",
"count": { "$sum": 1 }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 }
]).toArray();
// Run parallel top books for each addr
let topbooks = await Promise.all(
topaddr.map(({ _id: addr }) =>
books.aggregate([
{ "$match": { addr } },
{ "$group": {
"_id": "$book",
"count": { "$sum": 1 }
}},
{ "$sort": { "count": -1 } },
{ "$limit": 2 }
]).toArray()
)
);
// Merge output
topaddr = topaddr.map((d,i) => ({ ...d, books: topbooks[i] }));
log({ topaddr });
client.close();
} catch(e) {
console.error(e)
} finally {
process.exit()
}
})()
books.json
{ "addr": "address1", "book": "book1" }
{ "addr": "address2", "book": "book1" }
{ "addr": "address1", "book": "book5" }
{ "addr": "address3", "book": "book9" }
{ "addr": "address2", "book": "book5" }
{ "addr": "address2", "book": "book1" }
{ "addr": "address1", "book": "book1" }
{ "addr": "address15", "book": "book1" }
{ "addr": "address9", "book": "book99" }
{ "addr": "address90", "book": "book33" }
{ "addr": "address4", "book": "book3" }
{ "addr": "address5", "book": "book1" }
{ "addr": "address77", "book": "book11" }
{ "addr": "address1", "book": "book1" }
