Operasi agregasi di MongoDB memungkinkan Anda memproses catatan data, mengelompokkannya, dan mengembalikan hasil komputasinya. MongoDB mendukung tiga jenis operasi agregasi:

1. Perintah agregasi tujuan tunggal
2. Pengurangan Peta
3. Jalur agregasi

Anda dapat menggunakan dokumen perbandingan MongoDB ini untuk melihat mana yang sesuai dengan kebutuhan Anda.

Pipa Agregasi

Pipeline Agregasi adalah framework MongoDB yang menyediakan agregasi data melalui pipeline pemrosesan data. Artinya, dokumen dikirim melalui pipeline multi-langkah, memfilter, mengelompokkan, dan mengubah dokumen di setiap langkah. Ini menyediakan SQL "GROUP BY ...." jenis konstruksi untuk MongoDB yang berjalan di database itu sendiri. Dokumentasi agregasi memberikan contoh yang berguna untuk membuat saluran seperti itu.

Mengapa menjalankan agregasi di sekunder?

Pipeline agregasi adalah operasi yang intensif sumber daya – masuk akal untuk memindahkan tugas agregasi ke sekunder dari kumpulan replika MongoDB jika boleh beroperasi pada data yang sedikit usang. Ini biasanya berlaku untuk operasi 'batch' karena mereka tidak berharap untuk berjalan pada data terbaru. Jika output perlu ditulis ke koleksi, maka tugas agregasi hanya berjalan di primer karena hanya primer yang dapat ditulis di MongoDB.

Dalam postingan ini, kami akan menunjukkan kepada Anda cara memastikan bahwa agregasi pipeline dijalankan di secondary baik dari mongo shell dan Java.

Catatan:Kami menggunakan kumpulan data sampel yang disediakan oleh MongoDB dalam contoh agregasi kode pos mereka untuk menampilkan contoh kami. Anda dapat mengunduhnya seperti yang diinstruksikan dalam contoh.

Pipa Agregasi pada Kumpulan Replika

Cangkang MongoDB

Menyetel preferensi baca ke sekunder melakukan trik saat menjalankan pekerjaan agregasi dari mongo Shell. Mari kita coba mengambil semua negara bagian dengan populasi lebih dari 10 juta (agregasi pertama dalam contoh kode pos). Shell dan server menjalankan MongoDB versi 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 }

Melihat log MongoDB (dengan logging diaktifkan untuk perintah) di sekunder menunjukkan bahwa agregasi memang berjalan di sekunder:

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

Jawa

Dari driver Java MongoDB, sekali lagi pengaturan Preferensi baca berhasil. Berikut ini contoh menggunakan driver versi 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...");
    }
...
}

Masuk ke sekunder:

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

Tidak ada operasi yang direkam pada primer.

Pipa Agregasi pada Cluster yang Dibagikan

Pipeline agregasi didukung pada kluster yang di-shard. Perilaku terperinci dijelaskan dalam dokumentasi. Dari segi implementasi, ada sedikit perbedaan antara kumpulan replika dan sharded cluster saat menggunakan alur agregasi.

Cangkang MongoDB

Sebelum mengimpor data ke dalam sharding cluster, aktifkan sharding pada koleksi.

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

Setelah itu, operasinya sama dengan set replika:

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 }

Log dari salah satu sekunder:

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

Jawa

Kode yang sama seperti yang berlaku di set replika berfungsi dengan baik dengan kluster yang di-shard. Ganti saja string koneksi set replika dengan cluster yang di-sharded. Log dari sekunder menunjukkan bahwa pekerjaan memang dijalankan di sekunder:

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

Apakah konten ini bermanfaat? Beri tahu kami dengan men-tweet kami @scaledgridio dan seperti biasa, jika Anda memiliki pertanyaan, beri tahu kami di komentar di bawah. Oh dan! Jangan lupa untuk melihat produk hosting MongoDB kami yang dapat menghemat hingga 40% untuk biaya hosting MongoDB® jangka panjang.