Map-Reduce có lẽ là hoạt động linh hoạt nhất trong số các hoạt động tổng hợp mà MongoDB hỗ trợ.
Map-Reduce là một mô hình lập trình phổ biến bắt nguồn từ Google để xử lý và tổng hợp song song khối lượng lớn dữ liệu. Phần thảo luận chi tiết về Map-Reduce nằm ngoài phạm vi của bài viết này nhưng về cơ bản nó là một quá trình tổng hợp nhiều bước. Hai bước quan trọng nhất là giai đoạn bản đồ (xử lý từng tài liệu và đưa ra kết quả) và giai đoạn thu gọn (đối chiếu kết quả được phát ra trong giai đoạn bản đồ).
MongoDB hỗ trợ ba loại hoạt động tổng hợp:Bản đồ-Rút gọn, đường ống tổng hợp và các lệnh tổng hợp mục đích duy nhất. 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. Https://scalegrid.io/blog/mongodb-performance-running-mongodb-map-reduce-operations-on-secondaries/
Trong bài đăng cuối cùng của tôi, chúng ta đã thấy, với các ví dụ, cách chạy các đường ống Tổng hợp trên trang thứ hai. Trong bài đăng này, chúng ta sẽ hướng dẫn cách chạy các công việc Map-Reduce trên các bản sao phụ MongoDB.
MongoDB Map-Reduce
MongoDB hỗ trợ chạy các công việc Map-Reduce trên máy chủ cơ sở dữ liệu. Điều này mang lại sự linh hoạt để viết các tác vụ tổng hợp phức tạp không dễ thực hiện thông qua các đường ống tổng hợp. MongoDB cho phép bạn viết bản đồ tùy chỉnh và các hàm giảm thiểu trong Javascript có thể được chuyển đến cơ sở dữ liệu thông qua Mongo shell hoặc bất kỳ ứng dụng khách nào khác. Trên các tập dữ liệu lớn và không ngừng phát triển, người ta thậm chí có thể cân nhắc chạy các công việc Map-Reduce gia tăng để tránh xử lý dữ liệu cũ hơn mọi lúc.
Về mặt lịch sử, bản đồ và các phương thức rút gọn thường được thực thi trong ngữ cảnh đơn luồng. Tuy nhiên, hạn chế đó đã được loại bỏ trong phiên bản 2.4.
Tại sao phải chạy các công việc Map-Reduce trên Phụ?
Giống như các công việc tổng hợp khác, Map-Reduce cũng là một công việc 'hàng loạt' sử dụng nhiều tài nguyên, vì vậy nó rất phù hợp để chạy trên các bản sao chỉ đọc. Những lưu ý khi làm như vậy là:
1) Sẽ ổn nếu sử dụng dữ liệu hơi cũ. Hoặc bạn có thể điều chỉnh mối quan tâm ghi để đảm bảo các bản sao luôn đồng bộ với bản chính. Tùy chọn thứ hai này giả định rằng việc đánh vào hiệu suất ghi là có thể chấp nhận được.
2) Đầu ra của công việc Map-Reduce không được ghi vào một bộ sưu tập khác trong cơ sở dữ liệu mà phải được trả lại cho ứng dụng (tức là không ghi vào cơ sở dữ liệu).
Hãy xem cách thực hiện việc này thông qua các ví dụ, cả từ trình bao mongo và trình điều khiển Java.
Map-Reduce trên các tập bản sao
Tập dữ liệu
Để minh họa, chúng tôi sẽ sử dụng một tập dữ liệu khá đơn giản:Một kết xuất hồ sơ giao dịch hàng ngày từ một nhà bán lẻ. Một mục mẫu trông giống như:
RS-replica-0:PRIMARY> use test
switched to db test
RS-replica-0:PRIMARY> show tables
txns
RS-replica-0:PRIMARY> db.txns.findOne()
{
"_id" : ObjectId("584a3b71cdc1cb061957289b"),
"custid" : "cust_66",
"txnval" : 100,
"items" : [{"sku": sku1", "qty": 1, "pr": 100}, ...],
...
}
Trong các ví dụ của chúng tôi, chúng tôi sẽ tính toán tổng chi tiêu của một khách hàng nhất định vào ngày đó. Do đó, với lược đồ của chúng tôi, các phương thức bản đồ và phương thức thu gọn sẽ giống như sau:
var mapFunction = function() { emit(this.custid, this.txnval); } // Emit the custid and txn value from each record
var reduceFunction = function(key, values) { return Array.sum(values); } // Sum all the txn values for a given custid
đã cho Khi lược đồ của chúng tôi đã được thiết lập, hãy cùng xem hành động của Map-Reduce.
MongoDB Shell
Để đảm bảo rằng công việc Map-Reduce được thực thi trên phụ, tùy chọn đọc phải được đặt thành phụ . Giống như chúng tôi đã nói ở trên, để Map-Reduce chạy trên thiết bị phụ, đầu ra của kết quả phải là inline (Trên thực tế, đó là giá trị ngoại lệ duy nhất được phép trên giấy tờ thứ hai). Hãy xem nó hoạt động như thế nào.
$ mongo -u admin -p pwd --authenticationDatabase admin --host RS-replica-0/server-1.servers.example.com:27017,server-2.servers.example.com:27017
MongoDB shell version: 3.2.10
connecting to: RS-replica-0/server-1.servers.example.com:27017,server-2.servers.example.com:27017/test
2016-12-09T08:15:19.347+0000 I NETWORK [thread1] Starting new replica set monitor for server-1.servers.example.com:27017,server-2.servers.example.com:27017
2016-12-09T08:15:19.349+0000 I NETWORK [ReplicaSetMonitorWatcher] starting
RS-replica-0:PRIMARY> db.setSlaveOk()
RS-replica-0:PRIMARY> db.getMongo().setReadPref('secondary')
RS-replica-0:PRIMARY> db.getMongo().getReadPrefMode()
secondary
RS-replica-0:PRIMARY> var mapFunc = function() { emit(this.custid, this.txnval); }
RS-replica-0:PRIMARY> var reduceFunc = function(key, values) { return Array.sum(values); }
RS-replica-0:PRIMARY> db.txns.mapReduce(mapFunc, reduceFunc, {out: { inline: 1 }})
{
"results" : [
{
"_id" : "cust_0",
"value" : 72734
},
{
"_id" : "cust_1",
"value" : 67737
},
...
]
"timeMillis" : 215,
"counts" : {
"input" : 10000,
"emit" : 10000,
"reduce" : 909,
"output" : 101
},
"ok" : 1
}
Xem qua nhật ký trên thiết bị phụ xác nhận rằng công việc thực sự chạy trên thiết bị thứ cấp.
...
2016-12-09T08:17:24.842+0000 D COMMAND [conn344] mr ns: test.txns
2016-12-09T08:17:24.843+0000 I COMMAND [conn344] command test.$cmd command: listCollections { listCollections: 1, filter: { name: "txns" }, cursor: {} } keyUpdates:0 writeConflicts:0 numYields:0 reslen:150 locks:{ Global: { acquireCount: { r: 4 } }, Database: { acquireCount: { r: 1, R: 1 } }, Collection: { acquireCount: { r: 1 } } } protocol:op_query 0ms
2016-12-09T08:17:24.865+0000 I COMMAND [conn344] query test.system.js planSummary: EOF ntoreturn:0 ntoskip:0 keysExamined:0 docsExamined:0 cursorExhausted:1 keyUpdates:0 writeConflicts:0 numYields:0 nreturned:0 reslen:20 locks:{ Global: { acquireCount: { r: 6 } }, Database: { acquireCount: { r: 2, R: 1 } }, Collection: { acquireCount: { r: 2 } } } 0ms
2016-12-09T08:17:25.063+0000 I COMMAND [conn344] command test.txns command: mapReduce { mapreduce: "txns", map: function () { emit(this.custid, this.txnval); }, reduce: function (key, values) { return Array.sum(values); }, out: { inline: 1.0 } } planSummary: COUNT keyUpdates:0 writeConflicts:0 numYields:78 reslen:4233 locks:{ Global: { acquireCount: { r: 366 } }, Database: { acquireCount: { r: 3, R: 180 } }, Collection: { acquireCount: { r: 3 } } } protocol:op_command 220ms
...
Java
Bây giờ chúng ta hãy thử chạy công việc Map-Reduce trên các bản sao đã đọc từ một ứng dụng Java. Trên trình điều khiển Java MongoDB, thiết lập Tùy chọn đọc thực hiện một mẹo nhỏ. Đầu ra là nội tuyến theo mặc định nên không cần chuyển thêm tham số nào. 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 MapReduceExample {
private static final String MONGO_END_POINT = "mongodb://admin:[email protected]:27017,server-2.servers.example.com:27017/admin?replicaSet=RS-replica-0";
private static final String COL_NAME = "txns";
private static final String DEF_DB = "test";
public MapReduceExample() { }
public static void main(String[] args) {
MapReduceExample writer = new MapReduceExample();
writer.mapReduce();
}
public static final String mapfunction = "function() { emit(this.custid, this.txnval); }";
public static final String reducefunction = "function(key, values) { return Array.sum(values); }";
private void mapReduce() {
printer("Initializing...");
Builder options = MongoClientOptions.builder().readPreference(ReadPreference.secondary());
MongoClientURI uri = new MongoClientURI(MONGO_END_POINT, options);
MongoClient client = new MongoClient(uri);
MongoDatabase database = client.getDatabase(DEF_DB);
MongoCollection collection = database.getCollection(COL_NAME);
MapReduceIterable iterable = collection.mapReduce(mapfunction, reducefunction); // inline by default
MongoCursor cursor = iterable.iterator();
while (cursor.hasNext()) {
Document result = cursor.next();
printer("Customer: " + result.getString("_id") + ", Total Txn value: " + result.getDouble("value"));
}
printer("Done...");
}
...
}
Rõ ràng là từ các bản ghi, công việc chạy trên phụ:
...
2016-12-09T08:32:31.419+0000 D COMMAND [conn371] mr ns: test.txns
2016-12-09T08:32:31.420+0000 I COMMAND [conn371] command test.$cmd command: listCollections { listCollections: 1, filter: { name: "txns" }, cursor: {} } keyUpdates:0 writeConflicts:0 numYields:0 reslen:150 locks:{ Global: { acquireCount: { r: 4 } }, Database: { acquireCount: { r: 1, R: 1 } }, Collection: { acquireCount: { r: 1 } } } protocol:op_query 0ms
2016-12-09T08:32:31.444+0000 I COMMAND [conn371] query test.system.js planSummary: EOF ntoreturn:0 ntoskip:0 keysExamined:0 docsExamined:0 cursorExhausted:1 keyUpdates:0 writeConflicts:0 numYields:0 nreturned:0 reslen:20 locks:{ Global: { acquireCount: { r: 6 } }, Database: { acquireCount: { r: 2, R: 1 } }, Collection: { acquireCount: { r: 2 } } } 0ms
2016-12-09T08:32:31.890+0000 I COMMAND [conn371] command test.txns command: mapReduce { mapreduce: "txns", map: function() { emit(this.custid, this.txnval); }, reduce: function(key, values) { return Array.sum(values); }, out: { inline: 1 }, query: null, sort: null, finalize: null, scope: null, verbose: true } planSummary: COUNT keyUpdates:0 writeConflicts:0 numYields:156 reslen:4331 locks:{ Global: { acquireCount: { r: 722 } }, Database: { acquireCount: { r: 3, R: 358 } }, Collection: { acquireCount: { r: 3 } } } protocol:op_query 470ms
...
MongoDB Map-Reduce trên các cụm được chia nhỏ
MongoDB hỗ trợ Map-Reduce trên các cụm được phân đoạn, cả khi tập hợp được phân đoạn là đầu vào và khi nó là đầu ra của công việc Map-Reduce. Tuy nhiên, MongoDB hiện không hỗ trợ chạy các công việc thu nhỏ bản đồ trên các phân đoạn thứ hai của một cụm phân đoạn. Vì vậy, ngay cả khi tùy chọn out được đặt thành inline , Các công việc Map-Reduce sẽ luôn chạy trên các bản đồ chính của một cụm phân đoạn. Vấn đề này đang được theo dõi thông qua lỗi JIRA này.
Cú pháp của việc thực hiện công việc Map-Reduce trên một cụm phân đoạn cũng giống như cú pháp trên một tập hợp bản sao. Vì vậy, các ví dụ được cung cấp trong phần trên giữ nguyên. Nếu ví dụ Java ở trên được chạy trên một cụm phân đoạn, các thông báo nhật ký sẽ xuất hiện trên các thư mục sơ bộ cho biết rằng lệnh đã chạy ở đó.
...
2016-11-24T08:46:30.828+0000 I COMMAND [conn357] command test.$cmd command: mapreduce.shardedfinish { mapreduce.shardedfinish: { mapreduce: "txns", map: function() { emit(this.custid, this.txnval); }, reduce: function(key, values) { return Array.sum(values); }, out: { in
line: 1 }, query: null, sort: null, finalize: null, scope: null, verbose: true, $queryOptions: { $readPreference: { mode: "secondary" } } }, inputDB: "test", shardedOutputCollection: "tmp.mrs.txns_1479977190_0", shards: { Shard-0/primary.shard0.example.com:27017,secondary.shard0.example.com:27017: { result: "tmp.mrs.txns_1479977190_0", timeMillis: 123, timing: { mapTime: 51, emitLoop: 116, reduceTime: 9, mode: "mixed", total: 123 }, counts: { input: 9474, emit: 9474, reduce: 909, output: 101 }, ok: 1.0, $gleS
tats: { lastOpTime: Timestamp 1479977190000|103, electionId: ObjectId('7fffffff0000000000000001') } }, Shard-1/primary.shard1.example.com:27017,secondary.shard1.example.com:27017: { result: "tmp.mrs.txns_1479977190_0", timeMillis: 71, timing:
{ mapTime: 8, emitLoop: 63, reduceTime: 4, mode: "mixed", total: 71 }, counts: { input: 1526, emit: 1526, reduce: 197, output: 101 }, ok: 1.0, $gleStats: { lastOpTime: Timestamp 1479977190000|103, electionId: ObjectId('7fffffff0000000000000001') } } }, shardCounts: { Sha
rd-0/primary.shard0.example.com:27017,secondary.shard0.example.com:27017: { input: 9474, emit: 9474, reduce: 909, output: 101 }, Shard-1/primary.shard1.example.com:27017,secondary.shard1.example.com:27017: { inpu
t: 1526, emit: 1526, reduce: 197, output: 101 } }, counts: { emit: 11000, input: 11000, output: 202, reduce: 1106 } } keyUpdates:0 writeConflicts:0 numYields:0 reslen:4368 locks:{ Global: { acquireCount: { r: 2 } }, Database: { acquireCount: { r: 1 } }, Collection: { acqu
ireCount: { r: 1 } } } protocol:op_command 115ms
2016-11-24T08:46:30.830+0000 I COMMAND [conn46] CMD: drop test.tmp.mrs.txns_1479977190_0
...
Vui lòng truy cập trang sản phẩm MongoDB của chúng tôi để tìm hiểu về danh sách tính năng mở rộng của chúng tôi.
