Các hoạt động tổng hợp trong MongoDB cho phép bạn xử lý các bản ghi dữ liệu, nhóm chúng và trả về kết quả đã tính toán của chúng. MongoDB hỗ trợ ba loại hoạt động tổng hợp:

1. Các lệnh tổng hợp mục đích duy nhất
2. Thu nhỏ bản đồ
3. Quy trình tổng hợp

Bạn có thể sử dụng tài liệu so sánh MongoDB này để xem tài liệu nào phù hợp với nhu cầu của mình.

Đường ống tổng hợp

Đường ống tổng hợp là một khuôn khổ MongoDB cung cấp cho việc tổng hợp dữ liệu thông qua đường ống xử lý dữ liệu. Đó là tài liệu được gửi thông qua một đường dẫn nhiều bước, lọc, nhóm và chuyển đổi tài liệu ở mỗi bước. Nó cung cấp SQL “GROUP BY….” loại cấu trúc cho MongoDB chạy trên chính cơ sở dữ liệu. Tài liệu tổng hợp cung cấp các ví dụ hữu ích về việc tạo các đường ống như vậy.

Tại sao phải chạy tổng hợp trên trang phụ?

Các đường ống tổng hợp là các hoạt động sử dụng nhiều tài nguyên - việc giảm tải các công việc tổng hợp xuống cấp thứ hai của bộ bản sao MongoDB là hợp lý khi hoạt động trên dữ liệu hơi cũ là điều hợp lý. Điều này thường đúng đối với các hoạt động 'hàng loạt' vì chúng không mong đợi chạy trên dữ liệu mới nhất. Nếu đầu ra cần được ghi vào một bộ sưu tập thì các công việc tổng hợp chỉ chạy trên phần chính vì chỉ phần chính mới có thể ghi được trong MongoDB.

Trong bài đăng này, chúng tôi sẽ chỉ cho bạn cách đảm bảo rằng các đường ống tổng hợp được thực thi trên phần mềm phụ cả từ mongo shell và Java.

Lưu ý:Chúng tôi sử dụng tập dữ liệu mẫu do MongoDB cung cấp trong ví dụ tổng hợp mã zip của họ để giới thiệu các ví dụ của chúng tôi. Bạn có thể tải xuống theo hướng dẫn trong ví dụ.

Đường ống tổng hợp trên bộ bản sao

MongoDB shell

Đặt tùy chọn đọc thành phụ thực hiện thủ thuật khi chạy một công việc tổng hợp từ trình bao mongo. Hãy thử tìm nạp tất cả các tiểu bang có dân số lớn hơn 10 triệu (tập hợp thứ nhất trong ví dụ về mã zip). Cả trình bao và máy chủ đều đang chạy MongoDB phiên bản 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 }

Xem xét nhật ký MongoDB (có bật ghi nhật ký cho các lệnh) trên thứ cấp cho thấy rằng tổng hợp thực sự chạy trên thứ cấp:

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

Từ trình điều khiển Java MongoDB, một lần nữa thiết lập Tùy chọn đọc sẽ thực hiện thủ thuật. Dưới đây là một ví dụ sử dụng phiên bản trình điều khiển 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...");
    }
...
}

Nhật ký trên phụ:

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

Không có hoạt động nào được ghi lại trên chính.

Đường ống tổng hợp trên các cụm bị chia nhỏ

Các đường ống tổng hợp được hỗ trợ trên các cụm phân đoạn. Hành vi chi tiết được giải thích trong tài liệu. Về mặt triển khai, có rất ít sự khác biệt giữa tập hợp bản sao và cụm phân đoạn khi sử dụng đường dẫn tổng hợp.

MongoDB shell

Trước khi nhập dữ liệu vào cụm phân đoạn, hãy bật tính năng làm sắc nét trên bộ sưu tập.

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

Sau đó, các hoạt động giống như tập hợp bản sao:

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 }

Nhật ký từ một trong các sổ thứ hai:

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

Mã tương tự như có thể áp dụng trong tập hợp bản sao hoạt động tốt với một cụm phân đoạn. Chỉ cần thay thế chuỗi kết nối tập hợp bản sao bằng chuỗi kết nối được phân đoạn. Nhật ký từ thiết bị phụ chỉ ra rằng công việc thực sự đã được chạy trên thiết bị thứ hai:

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

Nội dung này có hữu ích không? Hãy cho chúng tôi biết bằng cách tweet tại chúng tôi @scaledgridio và như mọi khi, nếu bạn có bất kỳ câu hỏi nào, hãy cho chúng tôi biết trong phần bình luận bên dưới. Ồ và! Đừng quên xem các sản phẩm lưu trữ MongoDB của chúng tôi có thể tiết kiệm tới 40% chi phí lưu trữ MongoDB® dài hạn.