Met aggregatiebewerkingen in MongoDB kunt u gegevensrecords verwerken, groeperen en hun berekende resultaten retourneren. MongoDB ondersteunt drie soorten aggregatiebewerkingen:

1. Aggregatieopdrachten voor één doel
2. Kaart verkleinen
3. Aggregatiepijplijn

U kunt dit MongoDB-vergelijkingsdocument gebruiken om te zien welke bij uw behoeften past.

Aggregatiepijplijn

De aggregatiepijplijn is een MongoDB-framework dat voorziet in gegevensaggregatie via een gegevensverwerkingspijplijn. Dat wil zeggen dat documenten via een pijplijn met meerdere stappen worden verzonden, waarbij de documenten bij elke stap worden gefilterd, gegroepeerd en anderszins getransformeerd. Het biedt SQL “GROUP BY ….” type constructies voor MongoDB die op de database zelf worden uitgevoerd. Aggregatiedocumentatie biedt handige voorbeelden van het maken van dergelijke pijplijnen.

Waarom aggregaties uitvoeren op de secundaire?

Aggregatiepijplijnen zijn resource-intensieve bewerkingen. Het is logisch om aggregatietaken te verplaatsen naar secundaire onderdelen van een MongoDB-replicaset als het oké is om met enigszins verouderde gegevens te werken. Dit geldt meestal voor 'batch'-bewerkingen, omdat ze niet verwachten dat ze op de nieuwste gegevens worden uitgevoerd. Als de uitvoer naar een verzameling moet worden geschreven, worden de aggregatietaken alleen op de primaire uitgevoerd, omdat alleen de primaire in MongoDB kan worden geschreven.

In dit bericht laten we u zien hoe u ervoor kunt zorgen dat aggregatiepijplijnen op de secundaire worden uitgevoerd, zowel vanuit de mongo-shell als vanuit Java.

Opmerking:we gebruiken de voorbeeldgegevensset van MongoDB in hun voorbeeld van postcode-aggregatie om onze voorbeelden te laten zien. Je kunt het downloaden volgens de instructies in het voorbeeld.

Aggregatiepijplijn op replicasets

MongoDB-shell

De leesvoorkeur instellen op secundair doet de truc bij het uitvoeren van een aggregatietaak vanuit de mongo-shell. Laten we proberen alle staten op te halen met een bevolking van meer dan 10 miljoen (1e aggregatie in het voorbeeld van postcodes). Zowel de shell als de server draaien MongoDB versie 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 }

Een blik in de MongoDB-logboeken (met logboekregistratie ingeschakeld voor opdrachten) op de secundaire laat zien dat aggregatie inderdaad op de secundaire heeft plaatsgevonden:

...
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

Vanuit het MongoDB Java-stuurprogramma is het opnieuw instellen van de leesvoorkeur voldoende. Hier is een voorbeeld met driverversie 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...");
    }
...
}

Logt in op de secundaire:

...
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
...

Er is geen bewerking opgenomen op de primaire.

Aggregatiepijplijn op Sharded Clusters

Aggregatiepijplijnen worden ondersteund op shardclusters. Gedetailleerd gedrag wordt uitgelegd in de documentatie. Qua implementatie is er weinig verschil tussen replicaset en sharded-cluster bij gebruik van een aggregatiepijplijn.

MongoDB-shell

Schakel sharding voor de verzameling in voordat u gegevens in het shard-cluster importeert.

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

Daarna zijn de bewerkingen hetzelfde als de replicaset:

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 }

Logs van een van de secundairen:

...
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

Dezelfde code als die van toepassing is in de replicaset, werkt prima met een Sharded-cluster. Vervang gewoon de replicaset-verbindingsreeks door die van het Shard-cluster. Logboeken van een secundaire geven aan dat de taak inderdaad op de secundairen is uitgevoerd:

...
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
...

Was deze inhoud nuttig? Laat het ons weten door naar ons te tweeten @scaledgridio en zoals altijd, als je vragen hebt, laat het ons dan weten in de reacties hieronder. Oh en! Vergeet niet onze MongoDB-hostingproducten te bekijken waarmee u tot 40% kunt besparen op MongoDB®-hostingkosten op lange termijn.