MongoDB में एकत्रीकरण संचालन आपको डेटा रिकॉर्ड को संसाधित करने, उन्हें समूहबद्ध करने और उनके परिकलित परिणाम वापस करने की अनुमति देता है। MongoDB तीन प्रकार के एकत्रीकरण कार्यों का समर्थन करता है:

1. एकल उद्देश्य एकत्रीकरण आदेश
2. नक्शा-कम करें
3. एकत्रीकरण पाइपलाइन

आप इस MongoDB तुलना दस्तावेज़ का उपयोग यह देखने के लिए कर सकते हैं कि आपकी आवश्यकताओं के अनुरूप कौन सा है।

एकत्रीकरण पाइपलाइन

एकत्रीकरण पाइपलाइन एक MongoDB ढांचा है जो डेटा प्रोसेसिंग पाइपलाइन के माध्यम से डेटा एकत्रीकरण प्रदान करता है। यानी दस्तावेज़ एक बहु-चरण पाइपलाइन के माध्यम से भेजे जाते हैं, फ़िल्टरिंग, समूहीकरण और अन्यथा प्रत्येक चरण में दस्तावेज़ों को परिवर्तित करना। यह एसक्यूएल "ग्रुप बाय ..." प्रदान करता है। MongoDB के लिए प्रकार के निर्माण जो डेटाबेस पर ही चलते हैं। एकत्रीकरण प्रलेखन ऐसी पाइपलाइन बनाने के उपयोगी उदाहरण प्रदान करता है।

सेकेंडरी पर एग्रीगेशन क्यों चलाएं?

एकत्रीकरण पाइपलाइन संसाधन गहन संचालन हैं - जब थोड़े पुराने डेटा पर काम करना ठीक हो तो MongoDB प्रतिकृति सेट की सेकेंडरी में एकत्रीकरण नौकरियों को ऑफ़लोड करना समझ में आता है। यह आमतौर पर 'बैच' संचालन के लिए सच है क्योंकि वे नवीनतम डेटा पर चलने की उम्मीद नहीं करते हैं। यदि आउटपुट को किसी संग्रह में लिखने की आवश्यकता है तो एकत्रीकरण कार्य केवल प्राथमिक पर चलते हैं क्योंकि MongoDB में केवल प्राथमिक ही लिखने योग्य है।

इस पोस्ट में, हम आपको दिखाएंगे कि कैसे सुनिश्चित किया जाए कि मोंगो शेल और जावा दोनों से सेकेंडरी पर एग्रीगेशन पाइपलाइनों को निष्पादित किया जाए।

नोट:हम अपने उदाहरणों को प्रदर्शित करने के लिए MongoDB द्वारा प्रदान किए गए नमूना डेटा सेट का उपयोग उनके ज़िपकोड एकत्रीकरण उदाहरण में करते हैं। उदाहरण के निर्देशानुसार आप इसे डाउनलोड कर सकते हैं।

प्रतिकृति सेट पर एकत्रीकरण पाइपलाइन

MongoDB शेल

पठन वरीयता को माध्यमिक . पर सेट करना मोंगो खोल से एकत्रीकरण नौकरी चलाते समय चाल चलती है। आइए 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
...

जावा

MongoDB जावा ड्राइवर से, फिर से पठन वरीयता सेट करना चाल है। यहां ड्राइवर संस्करण 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
...

जावा

प्रतिकृति सेट में लागू समान कोड एक शार्प क्लस्टर के साथ ठीक काम करता है। बस प्रतिकृति सेट कनेक्शन स्ट्रिंग को शार्प किए गए क्लस्टर के साथ बदलें। द्वितीयक के लॉग इंगित करते हैं कि कार्य वास्तव में सेकेंडरी पर चलाया गया था:

...
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 होस्टिंग उत्पादों की जाँच करना न भूलें जो लंबी अवधि के MongoDB® होस्टिंग लागतों पर 40% तक बचा सकते हैं।