mongodb 分片配置以及mapreduce

淡淡的烟草味﹌ 2022-08-20 04:17 167阅读 0赞

三个mongod 其中20001为configdb

20002,20003为数据部分

27017为mongos部分

/usr/local/mongodb/bin/mongod --dbpath=/var/mongodb/data --logpath /var/mongodb/logs/log.log --port 20001 --nojournal --fork
    /usr/local/mongodb/bin/mongod --dbpath=/var/mongodb/data_3 --logpath /var/mongodb/logs/log_3.log --port 20002 --fork --nojournal
    /usr/local/mongodb/bin/mongod --dbpath=/var/mongodb/data_4 --logpath /var/mongodb/logs/log_4.log --port 20003 --fork --nojournal
    /usr/local/mongodb/bin/mongos --port 27017 --configdb=127.0.0.1:20001 --fork --logpath /var/mongodb/logs/log_2.log
    
    
    登录mongos后台
    /usr/local/mongodb/bin/mongo 127.0.0.1:27017/admin
    添加分片mongod
    db.runCommand({"addshard":"127.0.0.1:20002",allowLocal:true});
    db.runCommand({"addshard":"127.0.0.1:20003",allowLocal:true});
    开启数据分片
    db.runCommand({"enablesharding":"test"});
    db.runCommand({"shardcollection":"test.person","key":{"name":1}});

测试分片  
插入数据：

for(var i = 100;i<1000000;i++){ db.person.insert({"_id":i,"name":"terry"}) }

插入完成后：

db.printShardingStatus();

结果：

mongos> db.printShardingStatus();
    --- Sharding Status ---
      sharding version: {
        "_id" : 1,
        "version" : 4,
        "minCompatibleVersion" : 4,
        "currentVersion" : 5,
        "clusterId" : ObjectId("56ab0fc6b45916f958412ae6")
    }
      shards:
        {  "_id" : "shard0000",  "host" : "127.0.0.1:20002" }
        {  "_id" : "shard0001",  "host" : "127.0.0.1:20003" }
      databases:
        {  "_id" : "admin",  "partitioned" : false,  "primary" : "config" }
        {  "_id" : "test",  "partitioned" : true,  "primary" : "shard0000" }
            test.account
                shard key: { "name" : 1 }
                chunks:
                    shard0000    1
                { "name" : { "$minKey" : 1 } } -->> { "name" : { "$maxKey" : 1 } } on : shard0000 Timestamp(1, 0)
            test.account2
                shard key: { "_id" : 1 }
                chunks:
                    shard0001    3
                    shard0000    2
                { "_id" : { "$minKey" : 1 } } -->> { "_id" : NumberLong(1) } on : shard0001 Timestamp(2, 0)
                { "_id" : NumberLong(1) } -->> { "_id" : 11124 } on : shard0001 Timestamp(3, 0)
                { "_id" : 11124 } -->> { "_id" : 366409 } on : shard0000 Timestamp(4, 1)
                { "_id" : 366409 } -->> { "_id" : 725752 } on : shard0000 Timestamp(3, 2)
                { "_id" : 725752 } -->> { "_id" : { "$maxKey" : 1 } } on : shard0001 Timestamp(4, 0)
            test.person
                shard key: { "name" : 1 }
                chunks:
                    shard0000    1
                { "name" : { "$minKey" : 1 } } -->> { "name" : { "$maxKey" : 1 } } on : shard0000 Timestamp(1, 0)

这是分片的具体情况

{ "_id" : { "$minKey" : 1 } } -->> { "_id" : NumberLong(1) } on : shard0001 Timestamp(2, 0)
                { "_id" : NumberLong(1) } -->> { "_id" : 11124 } on : shard0001 Timestamp(3, 0)
                { "_id" : 11124 } -->> { "_id" : 366409 } on : shard0000 Timestamp(4, 1)
                { "_id" : 366409 } -->> { "_id" : 725752 } on : shard0000 Timestamp(3, 2)
                { "_id" : 725752 } -->> { "_id" : { "$maxKey" : 1 } } on : shard0001 Timestamp(4, 0)

mapreduce: mapreduce1.js

开始插入数据 插入的数据表为  分片表

/usr/local/mongodb/bin/mongo  mapreduce1.js

写一个mapreduce：

test_db = connect("localhost:27017/test");
    // mapreduce 
    map = function() {  
       emit(this.num, {
    		num:this.num,
    		count:1
    	   
       });  
    } ;
    
    reduce =    function(key, emits) {  
    		count = 0;
           for(x in emits){
    			count += emits[x].count;
    			num = emits[x].num;
    	   }
    		return {
    			num:num,
    			count:count
    		};
        }  
    	
    	
    finalize =	function(key, reducedValue) {  
       reducedValue.d = 1;
       return reducedValue;  
    }
    	
    	
    	
    test_db.mapreduce.mapReduce( 
    			map,  
                reduce,  
                {  
                    out: { 
    					merge: "map_reduce_example",
    					sharded:true
    				},  
                    query: { _id:{ $gt: 100000 } },  
                    finalize: finalize  
                }  
        );

/usr/local/mongodb/bin/mongo  mapreduce2.js

执行完登录后台

/usr/local/mongodb/bin/mongo 127.0.0.1:27017/test

db.printShardingStatus();

test.map_reduce_example
    			shard key: { "_id" : 1 }
    			chunks:
    				shard0000	1
    			{ "_id" : { "$minKey" : 1 } } -->> { "_id" : { "$maxKey" : 1 } } on : shard0000 Timestamp(1, 0) 
    		test.mapreduce
    			shard key: { "_id" : 1 }
    			chunks:
    				shard0000	2
    				shard0001	3
    			{ "_id" : { "$minKey" : 1 } } -->> { "_id" : 0 } on : shard0000 Timestamp(4, 0) 
    			{ "_id" : 0 } -->> { "_id" : 13676 } on : shard0000 Timestamp(3, 1) 
    			{ "_id" : 13676 } -->> { "_id" : 403345 } on : shard0001 Timestamp(4, 1) 
    			{ "_id" : 403345 } -->> { "_id" : 804273 } on : shard0001 Timestamp(4, 2) 
    			{ "_id" : 804273 } -->> { "_id" : { "$maxKey" : 1 } } on : shard0001 Timestamp(4, 3)

输入表和输出表如图

关键点为：

表在写入前先定义分区表

admindb.runCommand({"shardcollection":"test.mapreduce","key":{"_id":1}});

在out里面定义输出表为分区表：

test_db.mapreduce.mapReduce( 
    			map,  
                reduce,  
                {  
                    out: { 
    					merge: "map_reduce_example",
    					sharded:true
    				},  
                    query: { _id:{ $gt: 100000 } },  
                    finalize: finalize  
                }  
        );

在out中设置sharded为true即可

移除分片：

use admin;
    
     db.printShardingStatus();

查看分片状态，找到要删除的分片的名字，然后执行

mongos> db.runCommand({"removeshard":"shard0002"});
    {
        "msg" : "removeshard completed successfully",
        "state" : "completed",
        "shard" : "shard0002",
        "ok" : 1
    }

查看执行完成后的分片状态

db.printShardingStatus();

删除分片后，数据被拷贝到其他的分片中。