Операциите за агрегиране в MongoDB ви позволяват да обработвате записи с данни, да ги групирате и да връщате техните изчислени резултати. MongoDB поддържа три вида операции за агрегиране:

1. Команди за агрегиране с една цел
2. Карта-Намаляване
3. Тръбопровод за агрегиране

Можете да използвате този документ за сравнение на MongoDB, за да видите кой отговаря на вашите нужди.

Тръбопровод за агрегиране

Конвейерът за агрегиране е рамка на MongoDB, която осигурява агрегиране на данни чрез конвейер за обработка на данни. Това означава, че документите се изпращат чрез многоетапен тръбопровод, филтриране, групиране и по друг начин трансформиране на документите на всяка стъпка. Той предоставя SQL „GROUP BY….“ тип конструкции за MongoDB, които се изпълняват в самата база данни. Документацията за агрегиране предоставя полезни примери за създаване на такива тръбопроводи.

Защо да изпълнявате агрегирания на вторичния?

Конвейерите за агрегиране са ресурсоемки операции – има смисъл да се разтоварват задачи за агрегиране към вторични елементи на набор от реплики на MongoDB, когато е добре да работите с леко остарели данни. Това обикновено важи за „пакетни“ операции, тъй като те не очакват да работят с най-новите данни. Ако изходът трябва да бъде записан в колекция, тогава заданията за агрегиране се изпълняват само на основния, тъй като само първичният може да се записва в MongoDB.

В тази публикация ще ви покажем как да гарантирате, че тръбопроводите за агрегиране се изпълняват на вторичния елемент както от mongo shell, така и от Java.

Забележка:Използваме примерния набор от данни, предоставен от MongoDB в техния пример за агрегиране на пощенски кодове, за да покажем нашите примери. Можете да го изтеглите, както е указано в примера.

Тръбопровод за агрегиране върху набори реплики

Обвивка на MongoDB

Задаване на предпочитанието за четене на вторично прави трика, когато изпълнява задача за агрегиране от mongo shell. Нека се опитаме да извлечем всички държави с население над 10 милиона (първо обобщение в примера за пощенски кодове). И обвивката, и сървърът работят с MongoDB версия 3.2.10.

mongo -u admin -p <pwd> --authenticationDatabase admin --host RS-repl0-0/server-1.servers.example.com:27017,server-2.servers.example.com:27017
RS-repl0-0:PRIMARY> use test
switched to db test
RS-repl0-0:PRIMARY> db.setSlaveOk() // Ok to run commands on a slave
RS-repl0-0:PRIMARY> db.getMongo().setReadPref('secondary') // Set read pref
RS-repl0-0:PRIMARY> db.getMongo().getReadPrefMode()
secondary
RS-repl0-0:PRIMARY> db.zips.aggregate( [
...    { $group: { _id: "$state", totalPop: { $sum: "$pop" } } },
...    { $match: { totalPop: { $gte: 10*1000*1000 } } }
... ] )
{ "_id" : "CA", "totalPop" : 29754890 }
{ "_id" : "FL", "totalPop" : 12686644 }
{ "_id" : "PA", "totalPop" : 11881643 }
{ "_id" : "NY", "totalPop" : 17990402 }
{ "_id" : "OH", "totalPop" : 10846517 }
{ "_id" : "IL", "totalPop" : 11427576 }
{ "_id" : "TX", "totalPop" : 16984601 }

Преглед в регистрационните файлове на MongoDB (с активирано регистриране за команди) на вторичния елемент показва, че агрегирането наистина се изпълнява на вторичния:

...
2016-12-05T06:20:14.783+0000 I COMMAND  [conn200] command test.zips command: aggregate { aggregate: "zips", pipeline: [ { $group: { _id: "$state", totalPop: { $sum: "$pop" } } }, { 
$match: { totalPop: { $gte: 10000000.0 } } } ], cursor: {} } keyUpdates:0 writeConflicts:0 numYields:229 reslen:338 locks:{ Global: { acquireCount: { r: 466 } }, Database: { acquire
Count: { r: 233 } }, Collection: { acquireCount: { r: 233 } } } protocol:op_command 49ms
...

Java

От драйвера на MongoDB Java, отново настройката на Read Preference прави номера. Ето пример за използване на драйвер версия 3.2.2:

public class AggregationChecker {

    /*
     * Data and code inspired from:
     * https://docs.mongodb.com/v3.2/tutorial/aggregation-zip-code-data-set/#return-states-with-populations-above-10-million
     */
    private static final String MONGO_END_POINT = "mongodb://admin:[email protected]:27017,server-2.servers.example.com:27017/admin?replicaSet=RS-repl0-0";

    private static final String COL_NAME = "zips";
    private static final String DEF_DB = "test";

    public AggregationChecker() {
    }

    public static void main(String[] args) {
        AggregationChecker writer = new AggregationChecker();
        writer.aggregationJob();
    }

    private void aggregationJob() {
        printer("Initializing...");
        Builder options = MongoClientOptions.builder().readPreference(ReadPreference.secondary());
        MongoClientURI uri = new MongoClientURI(MONGO_END_POINT, options);
        MongoClient client = new MongoClient(uri);
        try {
            final DB db = client.getDB(DEF_DB);
            final DBCollection coll = db.getCollection(COL_NAME);
            // Avg city pop by state: https://docs.mongodb.com/manual/tutorial/aggregation-zip-code-data-set/#return-average-city-population-by-state
            Iterable iterable = coll.aggregate(
                    Arrays.asList(
                            new BasicDBObject("$group", new BasicDBObject("_id", new BasicDBObject("state", "$state").append("city", "$city")).append("pop",
                                    new BasicDBObject("$sum", "$pop"))),
                                    new BasicDBObject("$group", new BasicDBObject("_id", "$_id.state").append("avgCityPop", new BasicDBObject("$avg", "$pop"))))).results();

            for (DBObject entry : iterable) {
                printer(entry.toString());
            }
        } finally {
            client.close();
        }
        printer("Done...");
    }
...
}

Регистри на вторичния:

...
2016-12-01T10:54:18.667+0000 I COMMAND  [conn4113] command test.zips command: aggregate { aggregate: "zipcodes", pipeline: [ { $group: { _id: { state: "$state", city: "$city" }, pop: { $sum: "$pop" } } }, { $group: { _id: "$_id.state", avgCityPop: { $avg: "$pop" } } } ] } keyUpdates:0 writeConflicts:0 numYields:229 reslen:2149 locks:{ Global: { acquireCount: { r: 466 } }, Database: { acquireCount: { r: 233 } }, Collection: { acquireCount: { r: 233 } } } protocol:op_query 103ms
...

Не беше записана операция на основния.

Тръбопровод за агрегиране върху разчленени клъстери

Агрегационните тръбопроводи се поддържат на разчленени клъстери. Подробно поведение е обяснено в документацията. По отношение на внедряването има малка разлика между набора от реплики и раздробения клъстер, когато се използва конвейер за агрегиране.

Обвивка на MongoDB

Преди да импортирате данни в разделения клъстер, активирайте разделянето на колекцията.

mongos> sh.enableSharding("test")
mongos> sh.shardCollection("test.zips", { "_id" : "hashed" } )

След това операциите са същите като набора от реплика:

mongos> db.setSlaveOk()
mongos> db.getMongo().setReadPref('secondary')
mongos> db.getMongo().getReadPrefMode()
secondary
mongos> db.zips.aggregate( [
...    { $group: { _id: "$state", totalPop: { $sum: "$pop" } } },
...    { $match: { totalPop: { $gte: 10*1000*1000 } } }
... ] )
{ "_id" : "TX", "totalPop" : 16984601 }
{ "_id" : "PA", "totalPop" : 11881643 }
{ "_id" : "CA", "totalPop" : 29754890 }
{ "_id" : "FL", "totalPop" : 12686644 }
{ "_id" : "NY", "totalPop" : 17990402 }
{ "_id" : "OH", "totalPop" : 10846517 }
{ "_id" : "IL", "totalPop" : 11427576 }

Регистрации от един от вторичните:

...
2016-12-02T05:46:24.627+0000 I COMMAND  [conn242] command test.zips command: aggregate { aggregate: "zips", pipeline: [ { $group: { _id: "$state", totalPop: { $sum: "$pop" } } } ], fromRouter: true, cursor: { batchSize: 0 } } cursorid:44258973083 keyUpdates:0 writeConflicts:0 numYields:0 reslen:115 locks:{ Global: { acquireCount: { r: 4 } }, Database: { acquireCount: { r: 2 } }, Collection: { acquireCount: { r: 2 } } } protocol:op_query 0ms
2016-12-02T05:46:24.641+0000 I COMMAND  [conn131] getmore test.zips query: { aggregate: "zips", pipeline: [ { $group: { _id: "$state", totalPop: { $sum: "$pop" } } } ], fromRouter: true, cursor: { batchSize: 0 } } planSummary: PIPELINE_PROXY cursorid:44258973083 ntoreturn:0 keysExamined:0 docsExamined:0 cursorExhausted:1 keyUpdates:0 writeConflicts:0 numYields:112 nreturned:51 reslen:1601 locks:{ Global: { acquireCount: { r: 230 } }, Database: { acquireCount: { r: 115 } }, Collection: { acquireCount: { r: 115 } } } 13ms
...

Java

Същият код, както е приложим в набора от реплики, работи добре с разчленен клъстер. Просто заменете низа за връзка с набор от реплики с този на раздробения клъстер. Регистратурите от вторичен показват, че задачата наистина е била изпълнена на вторичните:

...
2016-12-02T05:39:12.339+0000 I COMMAND  [conn130] command test.zips command: aggregate { aggregate: "zips", pipeline: [ { $group: { _id: { state: "$state", city: "$city" }, pop: { $sum: "$pop" } } } ], fromRouter: true, cursor: { batchSize: 0 } } cursorid:44228970872 keyUpdates:0 writeConflicts:0 numYields:0 reslen:115 locks:{ Global: { acquireCount: { r: 4 } }, Database: { acquireCount: { r: 2 } }, Collection: { acquireCount: { r: 2 } } } protocol:op_query 0ms
2016-12-02T05:39:12.371+0000 I COMMAND  [conn131] getmore test.zips query: { aggregate: "zips", pipeline: [ { $group: { _id: { state: "$state", city: "$city" }, pop: { $sum: "$pop" } } } ], fromRouter: true, cursor: { batchSize: 0 } } planSummary: PIPELINE_PROXY cursorid:44228970872 ntoreturn:0 keysExamined:0 docsExamined:0 cursorExhausted:1 keyUpdates:0 writeConflicts:0 numYields:112 nreturned:12902 reslen:741403 locks:{ Global: { acquireCount: { r: 230 } }, Database: { acquireCount: { r: 115 } }, Collection: { acquireCount: { r: 115 } } } 30ms
...

Беше ли полезно това съдържание? Уведомете ни, като туитвате към нас @scaledgridio и както винаги, ако имате въпроси, уведомете ни в коментарите по-долу. О, и! Не забравяйте да разгледате нашите MongoDB хостинг продукти, които могат да спестят до 40% от дългосрочни разходи за хостинг на MongoDB®.