MongoDB使用小结：一些常用操作分享

[xxx]$ mongo --port 17380
MongoDB shell version: 2.4.5
connecting to: 127.0.0.1:17380/test
mongos> db.currentOp()
{ "inprog" : [ ...] }

mongos> db.killOp("shard0001:163415563")

db.currentOp().inprog.forEach(function(item){db.killOp(item.opid)})

db.currentOp().inprog.forEach(function(item){if(item.secs_running > 1000 )db.killOp(item.opid)})

db.currentOp().inprog.forEach(function(item){if(item.ns == "cswuyg.cswuyg")db.killOp(item.opid)})

db.currentOp().inprog.forEach(function(item){if(item.waitingForLock)print(JSON.stringify(item))})

db.currentOp().inprog.forEach(function(item){if(item.waitingForLock){print(item.opid.substr(0,9));print(item.op);}})

db.currentOp().inprog.forEach(function(item){if(!item.waitingForLock){var lock_info = item["opid"];print(lock_info.substr(0,9));print(item["op"]);}})

db.currentOp().inprog.forEach(function(item){if(item.op=="query"){print(item.opid);}})

db.currentOp().inprog.forEach(function(item){if(item.op!="query"){print(item.opid);}})

db.currentOp().inprog.forEach(function(item){print(item.op, item.opid);});

print("##########");db.currentOp().inprog.forEach(function(item){if(item.waitingForLock){var lock_info = item["opid"];print("waiting:",lock_info,item.op,item.ns);}});print("----");db.currentOp().inprog.forEach(function(item){if(!item.waitingForLock){var lock_info = item["opid"];print("doing",lock_info,item.op,item.ns);}});print("##########");

db.currentOp().inprog.forEach(function(item){if(item.waitingForLock){var lock_info = item["opid"];if(item.op=="query" && item.secs_running >60 && item.ns=="cswuyg.cswuyg"){db.killOp(item.opid)}}})

db.currentOp().inprog.forEach(function(item) {
 var lock_info = item["opid"];
 if (item.op == "query" && item.secs_running > 1000) {
  print("kill", item.opid);
  db.killOp(item.opid)
 }
})

{
        "cursor" : "BasicCursor",
        "isMultiKey" : false,
        "n" : 1,
        "nscannedObjects" : 10,
        "nscanned" : 10,
        "nscannedObjectsAllPlans" : 10,
        "nscannedAllPlans" : 10,
        "scanAndOrder" : false,
        "indexOnly" : false,
        "nYields" : 0,
        "nChunkSkips" : 0,
        "millis" : 0,
        "indexBounds" : {
        },
        "server" : "xxx:10001"
}

> db.runCommand({isdbgrid:1});
是：
{ "isdbgrid" : 1, "hostname" : "xxxhost", "ok" : 1 }
不是：
{
        "ok" : 0,
        "errmsg" : "no such cmd: isdbgrid",
        "code" : 59,
        "bad cmd" : {
                "isdbgrid" : 1
        }
}

看看当前的test collection上有啥索引：
mongos> db.test.getIndexes()
[
        {
                "v" : 1,
                "key" : {
                        "_id" : 1
                },
                "ns" : "cswuyg.test",
                "name" : "_id_"
        }
]
当前只有_id这个默认索引，我要在test collection上为 index 字段创建索引：
mongos> db.test.ensureIndex({"index": 1})
创建完了之后，再看看test collection上的索引有哪些：
mongos> db.test.getIndexes()
[
        {
                "v" : 1,
                "key" : {
                        "_id" : 1
                },
                "ns" : "cswuyg.test",
                "name" : "_id_"
        },
        {
                "v" : 1,
                "key" : {
                        "index" : 1
                },
                "ns" : "cswuyg.test",
                "name" : "index_1"
        }
]

mongos> db.a.ensureIndex({‘b‘: 1}, {‘unique‘: true})
{
        "raw" : {
                "set_a/xxxhost:20001,yyyhost:20002" : {
                        "createdCollectionAutomatically" : false,
                        "numIndexesBefore" : 1,
                        "ok" : 0,
                        "errmsg" : "cannot create unique index over { b: 1.0 } with shard key pattern { _id: 1.0 }",
                        "code" : 67
                },
                "set_b/xxxhost:30001,yyyhost:30002" : {
                        "createdCollectionAutomatically" : false,
                        "numIndexesBefore" : 1,
                        "ok" : 0,
                        "errmsg" : "cannot create unique index over { b: 1.0 } with shard key pattern { _id: 1.0 }",
                        "code" : 67
                }
        },
        "code" : 67,
        "ok" : 0,
        "errmsg" : "{ set_a/xxxhost:20001,yyyhos:20002: \"cannot create unique index over { b: 1.0 } with shard key pattern { _id: 1.0 }\", set_b/xxxhost:30001,yyyhost:30002: \"cannot create unique index over { b: 1.0 } with shard key pattern { _id: 1.0 }\" }"
}

//~/.mongorc.js
//show at begin
var compliment = ["attractive", "intelligent", "like batman"];
var index = Math.floor(Math.random()*3);
print("Hello, you‘re looking particularly " + compliment[index] + " today!");


//change the prompt
prompt = function(){
    if (typeof db == "undefined") {
        return "(nodb)> ";
    }
    // Check the last db operation
    try {
        db.runCommand({getLastError: 1});
    }
    catch (e) {
        print(e);
    }
    return db + "> ";
}


//show all shard‘s chunks
function my_show_shards() {
    var config_db = db.getSiblingDB("config");
    var collections = {};
    var shards = {};
    var shard_it = config_db.chunks.find().snapshot();

    while (shard_it.hasNext()) {
        next_item = shard_it.next();
        collections[JSON.stringify(next_item["ns"]).replace(/\"/g, "")] = 1;
        shards[JSON.stringify(next_item["shard"]).replace(/\"/g, "")] = 1;
    }
    var list_collections = [];
    var list_shards = [];
    for (item in collections) {
        list_collections.push(item);
    }
    for (item in shards) {
        list_shards.push(item);
    }

    list_collections.forEach(function(collec) {
            list_shards.forEach(function(item) {
                obj = {};
                obj["shard"] = item;
                obj["ns"] = collec;
                it = config_db.chunks.find(obj);
                print(collec, item, it.count());
                })
            })
}

db.chunks.find()

拷贝collection：

db.collection.copyTo("newcollection") 

36、聚合运算

包括：

1、pipeline；

2、map-reduce

管道：

http://docs.mongodb.org/manual/tutorial/aggregation-with-user-preference-data/

http://docs.mongodb.org/manual/reference/operator/aggregation/#aggregation-expression-operators

http://api.mongodb.org/python/current/api/pymongo/collection.html#pymongo.collection.Collection.aggregate

2.6之前的MongoDB，管道不支持超过16MB的返回集合。

可以使用$out操作符，把结果写入到collection中。如果aggregation成功，$out会替换已有的colleciton，但不会修改索引信息，如果失败，则什么都不做。

http://docs.mongodb.org/manual/reference/operator/aggregation/out/#pipe._S_out

38、修改Key:Value中的Value

给字段B的值加上大括号‘{‘:

db.test.find({_:ISODate("2014-11-02")}).forEach(function(item){if(/{.+}/.test(item["B"])){}else{print(item["B"]);db.test.update({"_id": item["_id"]}, {"$set": {"B": "{" + item["B"] + "}"}})}})

39、修改primary shard

db.runCommand({"movePrimary": "test", "to": "shard0000"})

这样子test DB 里的非sharded cocllection数据就会被存放到shard0000中，非空DB的话，可能会有一个migrate的过程。

40、 mongodb默认开启autobalancer

balancer是sharded集群的负载均衡工具，新建集群的时候默认开启，除非你在config里把它关闭掉：

config> db.settings.find()

{ "_id" : "chunksize", "value" : 64 }

{ "_id" : "balancer", "activeWindow" : { "start" : "14:00", "stop" : "19:30" }, "stopped" : false}

activeWindow指定autobalancer执行均衡的时间窗口。

stopped说明是否使用autobalancer。

手动启动balancer：sh.startBalancer()

判断当前balancer是否在跑：sh.isBalancerRunning()

41、MongoDB插入性能优化

插入性能：200W的数据，在之前没有排序就直接插入，耗时4小时多，现在，做了排序，插入只需要5分钟。排序对于单机版本的MongoDB性能更佳，避免了随机插入引发的频繁随机IO。

排序：在做分文件排序的时候，文件分得越小，排序越快，当然也不能小到1，否则频繁打开文件也耗费时间。

42、MongoDB数组操作

mongos> db.test.update({"helo":"he2"}, {"$push": {"name":"b"}})

多次插入后结果：

{ "_id" : ObjectId("54a7aa2be53662aebc28585f"), "helo" : "he2", "name" : [ "a", "b", "b" ] }

2、更新/插入数据，保证不重复：

mongos> db.test.update({"helo":"she"}, {"$addToSet": {"name":"b"}})

WriteResult({ "nMatched" : 1, "nUpserted" : 0, "nModified" : 0 })

多次插入后结果：

{ "_id" : ObjectId("54dd6e1b570cb10de814f86d"), "helo" : "she", "name" : [ "b" ] }

保证只有一个值。

3、数组元素个数：

$size 用来指定数组的元素个数，显示fruit数组长度为3的document：

mongos> db.a.find({"fruit": {$size: 3}})

{ "_id" : ObjectId("54b334798220cd3ad74db314"), "fruit" : [ "apple", "orange", "cherry" ] }

43、更换Key: Value中的Key

verRelease 换为 TEST

db.test.find().forEach(function(item){db.test.update({_id:item["_id"]}, {"$set": {"TEST": item["verRelease"]}}, {"$unset":{"verRelease":{"exists":true}}})})

或者更新部分修改为：

db.test.update({D :  ISODate("2014-11-02")}, {$rename :  {"AcT": "AM"}}, false, true)

config> sh.moveChunk("xx.yy", { "_id" : { "D" : ISODate("2015-02-24T00:00:00Z"), "id" : "3f" } }, "shard0003")

如果出现错误，参考这里：可能需要重启 

http://stackoverflow.com/questions/26640861/movechunk-failed-to-engage-to-shard-in-the-data-transfer-cant-accept-new-chunk

46、格式化json显示

db.collection.find().pretty()

47、更新replica集群的域名信息

cfg = rs.conf()
cfg.members[0].host = "xxxhost: 20000"
cfg.members[1].host = "yyyhost: 20001"
cfg.members[2].host = "zzzhost: 20002"
rs.reconfig(cfg)

48、不要直接修改local.system.replset

不要直接修改local.system.replset，因为他只能修改本机器的本地信息。

但是如果出现了：

 {
        "errmsg" : "exception: can‘t use localhost in repl set member names except when using it for all members",
        "code" : 13393,
        "ok" : 0
 }

这样的错误，那就要修改本机的local.system.replset，然后重启。

49、排重统计

（1）aggregate

 result = db.flu_test.aggregate([{$match: {_: ISODate("2015-05-01")}}, {$group:{_id:"$F", value: {$sum:1}}}], {allowDiskUse:true})

 result.itcount()

（2）distinct

非排重文档量统计：

mongos> count = db.flu_test.aggregate([{$match:{_:ISODate("2015-05-21")}}, {$group:{_id:null, value: {$sum:1}}}], {allowDiskUse:true})

{ "_id" : null, "value" : 3338987 }

50、pymongo优先读取副本集Secondary节点

优先读取副本集Secondary节点，可以减少primary节点负担，在primary节点跟secondary节点同步延迟较短、业务对数据不要求实时一致时可以利用副本集做读写分离和负载均衡。

副本集集群的读取有这几种使用方式：

primary: 默认参数，只从主节点读取；
primaryPreferred: 大部分从主节点上读取，主节点不可用时从Secondary节点读取；
secondary: 只从Secondary节点上进行读取操作；
secondaryPreferred: 优先从Secondary节点读取，Secondary节点不可用时从主节点读取；
nearest: 从网络延迟最低的节点上读取。

（1）测试1 优先从secondary读取数据:

import pymongo


client = pymongo.MongoReplicaSetClient(‘xxxhost: yyyport‘,  replicaSet=‘my_set‘, readPreference=‘secondaryPreferred‘)
print client.read_preference  # 显示当前的读取设定
for i in xrange(1, 10000):  # 循环10000次，用mongostat观察查询负载
    a = client[‘history‘][‘20140409‘].find_one({"ver_code": "128"})
    print a

（2）测试2 直接连接某Secondary节点读取数据：

import pymongo


client = pymongo.MongoClient(‘xxxhost‘, yyyport, slaveOk=True)
a = client[‘msgdc_ip‘][‘query_ip‘].find().count()
print a

http://www.lanceyan.com/category/tech/mongodb
http://emptysqua.re/blog/reading-from-mongodb-replica-sets-with-pymongo/

http://api.mongodb.org/python/current/api/pymongo/read_preferences.html#module-pymongo.read_preferences

http://api.mongodb.org/python/current/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient

注意：3.0之后MongoReplicaSetClient函数是要被放弃的。

但是测试时发现：在较低版本中，需要使用MongoReplicaSetClient，MongoClient无法实现 pymongo.ReadPreference.SECONDARY_PREFERRED功能。

2015.12.28补充：

51、为副本集设置标签

可以为副本集中的每个成员设置tag（标签），设置标签的好处是后面读数据时，应用可以指定从某类标签的副本上读数据。

为replica 设置tag
在副本shell下执行：
var conf = rs.conf()
conf.members[0].tags = { "location": "beijing" }
conf.members[1].tags = { "location": "hangzhou"}
conf.members[2].tags = { "location": "guangzhou"  }
rs.reconfig(conf)

 参考：https://docs.mongodb.org/v3.0/tutorial/configure-replica-set-tag-sets/ 

52、副本集碎片整理的一种方法

使用MMAPv1存储引擎时，对于频繁大数据量的写入删除操作，碎片问题会变得很严重。在数据同步耗时不严重的情况下，我们不需要对每个副本做repair，而是轮流“卸下副本，删除对应的磁盘文件，重新挂上副本”。每个重新挂上的副本都会自动去重新同步一遍数据，碎片问题就解决了。

53、存储引擎升级为wiredTiger

我们当前的版本是MongoDB3.0.6，没有开启wiredTiger引擎，现在打算升级到wiredTiger引擎。

我们是Shared Cluster，shard是Replica Set。升级比较简单，只需要逐步对每一个副本都执行存储引擎升级即可，不影响线上服务。

升级时，只在启动命令中添加：--storageEngine wiredTiger。

步骤：首先，下掉一个副本；然后，把副本的磁盘文件删除掉；接着，在该副本的启动命令中添加--storageEngine wiredTiger后启动。这就升级完一个副本，等副本数据同步完成之后，其它副本也照样操作（或者处理完一个副本之后，拷贝该副本磁盘文件替换掉另一个副本的磁盘文件）。风险：如果数据量巨大，且有建索引的需求，容易出现内存用尽。

分片集群的configure server可以不升级。

升级之后磁盘存储优化效果极度明显，22GB的数据会被精简到1.3GB。

升级后的磁盘文件完全变了，所以不同存储引擎下的磁盘文件不能混用。

升级参考：https://docs.mongodb.org/manual/tutorial/change-sharded-cluster-wiredtiger/

54、oplogSizeMB不要设置得太大

启动配置中的这个字段是为了设置oplog collection的大小，oplog是操作记录，它是一个capped collection，在副本集群中，设置得太小可能导致secondary无法及时从primary同步数据。默认情况下是磁盘大小的5%。但是，如果这个字段设置得太大，可能导致暴内存，oplog的数据几乎是完全加载在内存中，一旦太大，必然暴内存，导致OOM。而且因为这个collection是capped，MongoDB启动之后无法修改其大小。