MongoDB 介绍和操作

前言

什么是 MongoDB？ 

• MongoDB 是一个介于关系数据库和非关系数据库之间的开源产品，是最接近于关系型数据库的 NoSQL 数据库。
• 它在轻量级JSON 交换基础之上进行了扩展，即称为 BSON 的方式来描述其无结构化的数据类型。尽管如此它同样可以存储较为复杂的数据类型。
• 它和上一篇文章讲到的Redis有异曲同工之妙。虽然两者均为 NoSQL ，但是 MongoDB 相对于 Redis 而言，MongoDB 更像是传统的数据库。
• 早些年我们是先有了 Relation Database (关系型数据库)，然后出现了很多很复杂的query ，里面用到了很多嵌套，很多 join 操作。
• 所以在设计数据库的时候，我们也考虑到了如何应用他们的关系，使得写 query 可以使 database 效率达到最高。
• 后来人们发现，不是每个系统，都需要如此复杂的关系型数据库。有些简单的网站，比如博客，比如社交网站，完全可以斩断数据库之间的一切关系。
• 这样做带来的好处是，设计数据库变得更加简单，写 query 也变得更加简单。然后，query 消耗的时间可能也会变少。
• 因为 query 简单了，少了许多消耗资源的 join 操作，速度自然会上去。正如所说的， query 简单了，很有以前 MySQL 可以找到的东西，现在关系没了
• 通过 Mongo 找不到了。我们只能将几组数据都抓到本地，然后在本地做 join ，所以在这点上可能会消耗很多资源。这里我们可以发现。如何选择数据库
• 完全取决于你所需要处理的数据的模型，即 Data Model 。如果它们之间，关系错综复杂，千丝万缕，这个时候 MySQL 一定是首选。
• 如果他们的关系并不是那么密切，那么， NoSQL 将会是利器。

MongoDB 和 Redis 一样均为 key-value 存储系统，它具有以下特点：

• 面向集合存储，易存储对象类型的数据。 模式自由。 支持动态查询。 支持完全索引，包含内部对象。
• 支持查询。 支持复制和故障恢复。 使用高效的二进制数据存储，包括大型对象(如视频等)。
• 自动处理碎片，以支持云计算层次的扩展性 支持 Python ， PHP ， Ruby ， Java ， C ， C# ， Javascript ，Perl 及 C++ 语言的驱动程序
• 社区中也提供了对 Erlang 及 .NET 等平台的驱动程序。 文件存储格式为 BSON (一种 JSON 的扩展)。 可通过网络访问。
• 无模式（太过随意，有时反倒是缺点）
• 支持对象存储
• 支持Map/reduce和聚合操作
• 扩展方便
• 可靠性高
• MongoDB的缺点不多，但很要命，这就是被很多人诟病的“内存贪婪”：它会占用操作系统几乎所有的空闲内存
• 让其他进程活得不舒适，而我们一直对该机制缺乏了解，也没有相应的应对手段

MongoDB 与 MySQL 性能比较

• 像 MySQL 一样， MongoDB 提供了丰富的远远超出了简单的键值存储中提供的功能和功能。 MongoDB 具有查询语言，功能强大的辅助索引(包括文本搜索和地理空间)，数据分析功能强大的聚合框架等。相比使用关系数据库而言，使用MongoDB ，您还可以使用如下表所示的这些功能，跨越更多样化的数据类型和数据规模。
• MySQL MongoDB丰富的数据模型否是动态 Schema否是数据类型是是数据本地化否是字段更新是是易于编程否是复杂事务是否审计是是自动分片否是
• MySQL 中的许多概念在 MongoDB 中具有相近的类比。本表概述了每个系统中的一些常见概念。
• MySQL MongoDB表集合行文档列字段joins嵌入文档或者链接

MongoDB应用范围和限制

• MongoDB 的主要目标是在 key-value (键/值)存储方式(提供了高性能和高度伸缩性)以及传统的 RDBMS 系统(丰富的功能)架起一座桥梁，集两者的优势于一身。

MongoDB 适用范围如下：

• 网站数据： Mongo 非常适合实时的插入，更新与查询，并具备网站实时数据存储所需的复制及高度伸缩性。
• 缓存：由于性能很高， Mongo 也适合作为信息基础设施的缓存层。在系统重启之后，由 Mongo 搭建的持久化缓存层可以避免下层的数据源过载。
• 大尺寸，低价值的数据：使用传统的关系型数据库存储一些数据时可能会比较昂贵，在此之前，很多时候程序员往往会选择传统的文件进行存储。
• 高伸缩性的场景： Mongo 非常适合由数十或数百台服务器组成的数据库。 Mongo 的路线图中已经包含对 MapReduce 引擎的内置支持。
• 用于对象及 JSON 数据的存储： Mongo 的 BSON 数据格式非常适合文档化格式的存储及查询。MongoDB 当然也会有以下场景的限制
• 高度事物性的系统：例如银行或会计系统。传统的关系型数据库目前还是更适用于需要大量原子性复杂事务的应用程序。
• 传统的商业智能应用：针对特定问题的 BI 数据库会对产生高度优化的查询方式。对于此类应用，数据仓库可能是更合适的选择。 需要 SQL 的问题。

MongoDB客户端类

pymongo是python访问MongoDB的模块，使用该模块，我们定义了一个操作MongoDB的类PyMongoClient，包含了连接管理、集合管理、索引管理、增删改查、文件操作、聚合操作等方法。

# ---------------------------------------------------------------------------------------------------
# PyMongoClient(host='localhost', port='27017', db='test', user=None, passwd=None, loop=5, rate=8)
# ---------------------------------------------------------------------------------------------------

PyMongoClient.SetDatabase(db, user, passwd)  # 设置（修改）当前数据库
PyMongoClient.CloseConnection()              # 关闭连接
PyMongoClient.Logout()   #  注销用户
PyMongoClient.GetStatus()#  获取MogoDB服务器的状态
PyMongoClient.IsMongos() # 判断是否是MongoS
PyMongoClient.GetDateTime()#  获取MongoDB服务器的当前时间（需要权限支持）


# ---------------------------------------------------------------------------------------------------

PyMongoClient.GetCollections() # 获取当前数据库的全部集合
PyMongoClient.CreateCollection(collection)#  创建集合
PyMongoClient.DropCollection(collection)#  删除集合

# ---------------------------------------------------------------------------------------------------

PyMongoClient.IndexInformation(collection)#  获取集合的索引信息
PyMongoClient.EnsureIndex(collection, key_or_list)#  检查索引是否存在，若不存在，则创建索引
PyMongoClient.CreateIndex(collection, key_or_list)#  创建索引
PyMongoClient.DropIndex(collection, key=None)#  删除索引，key=None时删除全部索引（_id除外）

# ---------------------------------------------------------------------------------------------------

PyMongoClient.InsertDoc(collection, data)#  data为字典时，单条插入，data为列表时，批量插入
PyMongoClient.RemoveDoc(collection, docFilter=None)#  删除文档，docFilter=None时删除集合的全部文档
PyMongoClient.UpdateDoc(collection, docFilter, data, modifier=None)#  更新文档，支持使用$inc/$set/$unset等修改器
PyMongoClient.UpsertDoc(collection, docFilter, data)#  如果文档不存在，则插入文档；如果文档存在，则更新文档
PyMongoClient.GetDoc(collection, docFilter=None, colFilter=None)#  返回单个文档
PyMongoClient.CountDoc(collection, docFilter=None)#  返回集合或查询的文档总数
PyMongoClient.GetCursor(collection, docFilter=None, colFilter=None)#  返回多个文档的游标
PyMongoClient.CountCursor(cursor) # 返回游标的文档总数
PyMongoClient.SortCursor(cursor, col_or_list, director='ASC')#  游标排序，默认升序，取值ASC/DESC
PyMongoClient.SubCursor(cursor, limit, skip=0)#  游标截取

# ---------------------------------------------------------------------------------------------------

PyMongoClient.Aggregate(collection, pipleline) # 聚合
PyMongoClient.RunCommand(collection, cmdObj) # 运行数据库命令

# ---------------------------------------------------------------------------------------------------

PyMongoClient.Str2ObjectId(id_str)#  id字符串转id对象
PyMongoClient.ObjectId2Str(id_obj)#  id对象转id字符串
PyMongoClient.GetBinaryFromFile(sourceFile) # 读文件，返回二进制内容
PyMongoClient.SaveBinaryToFile(binary, targetFile) # 将二进制内容保存为文件

# ---------------------------------------------------------------------------------------------------

PyMongoClient.PutFile(localFilePath, dbFileName=None) # 将文件保存到GridFS并返回FileId
PyMongoClient.GetFile(fileId, localFilePath)#  将文件从GridFS取出，并保存到文件中
PyMongoClient.GetFilesCursor(docFilter=None, colFilter=None)#  取得文件信息游标
PyMongoClient.DeleteFile(fileId) # 删除GridFS中的文件

连接管理

class PyMongoClient():
def __init__(self, host='localhost', port=27017, db='test', user=None, passwd=None, loop=5, rate=8):
self.loop = loop # 数据库失去连接后，尝试执行数据库操作的次数
self.rate = float(rate) # 数据库失去连接后，尝试执行数据库操作的时间间隔，首次尝试的间隔是rate的倒数，以后间隔时间增倍
try:
self.conn = pymongo.MongoClient(host, int(port))
self.SetDatabase(db, user, passwd)
except Exception, errMsg:
raise Exception(errMsg)

# ---------------------------------------------------------------------------------------------------
def SetDatabase(self, db, user, passwd):
# 设置（修改）当前数据库
self.db = self.conn[db]
if user and passwd:
if not self.db.authenticate(user, passwd):
raise Exception(u'数据库权限验证失败！')

def CloseConnection(self):
# 关闭数据库连接
self.conn.close()

def Logout(self):
# 注销用户
self.db.logout()

def GetStatus(self):
# 获取MogoDB服务器的状态
return self.db.last_status()

def IsMongos(self):
# 判断是否是MongoS
return self.conn.is_mongos

def GetDateTime(self):
# 获取MongoDB服务器的当前时间（需要权限支持，若无权限，则返回本地时间）
for i in range(self.loop):
try:
return self.db.eval("return new Date();")
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
except Exception, e:
return datetime.datetime.now()

raise Exception(u'重连数据库失败！')

集合管理

class PyMongoClient():
def GetCollections(self):
# 获取当前数据库的全部集合
for i in range(self.loop):
try:
return self.db.collection_names()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def CreateCollection(self, collection):
# 在当前数据库内创建新的集合
for i in range(self.loop):
try:
self.db.create_collection(collection)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
except pymongo.errors.CollectionInvalid:
return
raise Exception(u'重连数据库失败！')

def DropCollection(self, collection):
# 删除当前数据库内名为collection的集合
for i in range(self.loop):
try:
self.db.drop_collection(collection)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

索引管理

class PyMongoClient():
def IndexInformation(self, collection):
# 获取索引信息
for i in range(self.loop):
try:
return self.db[collection].index_information().keys()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def EnsureIndex(self, collection, key_or_list):
# 检查索引是否存在，若不存在，则创建索引，若存在，返回None
# list参数形如：[('start_time', pymongo.ASCENDING), ('end_time', pymongo.ASCENDING)]
for i in range(self.loop):
try:
self.db[collection].ensure_index(key_or_list)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def CreateIndex(self, collection, key_or_list):
# 创建索引（推荐使用EnsureIndex）
for i in range(self.loop):
try:
self.db[collection].create_index(key_or_list)
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def DropIndex(self, collection, key=None):
# 删除索引，key=None时删除全部索引（_id除外）
for i in range(self.loop):
try:
if key:
self.db[collection].drop_index(key)
else:
self.db[collection].drop_indexes()
return
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

增删改查

class PyMongoClient():
def InsertDoc(self, collection, data):
# data为字典时，单条插入，data为列表时，批量插入。批量上限20W
# 单条插入时返回单个id对象，批量插入时，返回id对象列表
for i in range(self.loop):
try:
return self.db[collection].insert(data, manipulate=True, save=False, check_keys=True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def RemoveDoc(self, collection, docFilter=None):
# 删除文档，docFilter=None时删除集合collection的全部文档
for i in range(self.loop):
try:
return self.db[collection].remove(docFilter)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def UpdateDoc(self, collection, docFilter, data, modifier=None):
# 更新文档，docFilter为更新对象的查找条件，data为更新数据，可以使用$inc/$set/$unset等修改器
for i in range(self.loop):
try:
if modifier:
return self.db[collection].update(docFilter, {modifier:data}, multi=True)
else:
return self.db[collection].update(docFilter, data, multi=True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def UpsertDoc(self, collection, docFilter, data):
# 如果文档不存在，则插入文档；如果文档存在，则更新文档
for i in range(self.loop):
try:
return self.db[collection].update(docFilter, data, True)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def GetDoc(self, collection, docFilter=None, colFilter=None, sort=None):
# 返回单个文档
for i in range(self.loop):
try:
if colFilter:
return self.db[collection].find_one(docFilter, colFilter, sort=sort)
else:
return self.db[collection].find_one(docFilter, sort=sort)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def CountDoc(self, collection, docFilter=None):
# 返回集合或查询的文档总数
for i in range(self.loop):
try:
return self.db[collection].find(docFilter).count()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def GetCursor(self, collection, docFilter=None, colFilter=None):
# 返回多个文档的游标
for i in range(self.loop):
try:
if colFilter:
return self.db[collection].find(docFilter, colFilter).batch_size(100)
else:
return self.db[collection].find(docFilter).batch_size(100)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def CountCursor(self, cursor):
# 返回游标的文档总数
for i in range(self.loop):
try:
return cursor.count()
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def SortCursor(self, cursor, col_or_list, director='ASC'):
# 游标排序，默认ASCENDING（升序），取值ASC/DESC
# col_or_list，列名或者是由(列名,方向)组成的列表
if isinstance(col_or_list, list):
args = []
for col in col_or_list:
if col[1] == 'ASC':
args.append((col[0],pymongo.ASCENDING))
else:
args.append((col[0],pymongo.DESCENDING))
for i in range(self.loop):
try:
return cursor.sort(args) # cursor.sort([("UserName",pymongo.ASCENDING),("Email",pymongo.DESCENDING)])
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')
else:
if director == 'ASC':
director = pymongo.ASCENDING
else:
director = pymongo.DESCENDING
for i in range(self.loop):
try:
return cursor.sort(col_or_list, director) # director取值：pymongo.ASCENDING（升序）、pymongo.DESCENDING（降序）
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def SubCursor(self, cursor, limit, skip=0):
# 截取游标
for i in range(self.loop):
try:
if skip:
return cursor.skip(skip).limit(limit)
else:
return cursor.limit(limit)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

文件操作

class PyMongoClient():
def GetBinaryFromFile(self, sourceFile):
# 读文件，返回二进制内容
# 适用于在文档中直接保存小于16M的小文件，若文件较大时，应使用GridFS
try:
fp = open(sourceFile,'rb')
return bson.Binary(fp.read())
except:
return False
finally:
fp.close()

def SaveBinaryToFile(self, binary, targetFile):
# 将二进制内容保存为文件
try:
fp = open(targetFile,'wb')
fp.write(binary)
return True
except:
return False
finally:
fp.close()

def Str2ObjectId(self, id_str):
return bson.ObjectId(id_str)

def ObjectId2Str(self, id_obj):
return str(id_obj)

def PutFile(self, localFilePath, dbFileName=None):
'''
向GridFS中上传文件，并返回文件ID
@localFilePath 本地文件路径
@dbFileName 保存到GridFS中的文件名，如果为None则使用本地路径中的文件名
'''

fs = gridfs.GridFS(self.db)
fp = open(localFilePath, 'rb')
if dbFileName == None:
dbFileName = os.path.split(localFilePath)[1]
id = fs.put(fp,filename=dbFileName, chunkSize=4*1024*1024)
fp.close()
return id

def GetFile(self, fileId, localFilePath=None):
'''
根据文件ID从GridFS中下载文件
@fileId 文件ID
@localFilePath 要保存的本地文件路径
'''

if isinstance(fileId, str):
fileId = self.Str2ObjectId(fileId)

fs = gridfs.GridFS(self.db)
if localFilePath:
fp = open(localFilePath, 'wb')
try:
fp.write(fs.get(fileId).read())
return True
except:
return False
finally:
fp.close()
else:
try:
return fs.get(fileId).read()
except:
return False
def GetFilesCursor(self, docFilter=None, colFilter=None):
'''
取得GridFS中文件的游标
可以进行过滤或检索的字段名有
_id 文件ID
filename 文件名
length 文件大小
md5 md5校验码
chunkSize 文件块大小
uploadDate 更新时间
'''

return self.GetCursor('fs.files', docFilter=docFilter, colFilter=colFilter)

def DeleteFile(self, fileId):
'''
根据文件ID从GridFS中删除文件
@fileId 文件ID
'''

fs = gridfs.GridFS(self.db)
fs.delete(fileId)

聚合操作

class PyMongoClient():
def Aggregate(self, collection, pipleline):
# 聚合
# pipleline是一个由筛选、投射、分组、排序、限制、跳过等一系列构件组成管道队列
for i in range(self.loop):
try:
return self.db[collection].aggregate(pipleline)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

def RunCommand(self, collection, cmdObj):
# 运行数据库命令
# if cmdObj is a string, turns it into {cmdObj:1}
for i in range(self.loop):
try:
return self.db[collection].runCommand(cmdObj)
except pymongo.errors.AutoReconnect:
time.sleep(pow(2,i)/self.rate)
raise Exception(u'重连数据库失败！')

MongoDB使用场景

  更高的写入负载：

• 默认情况下，MongoDB更侧重高数据写入性能，而非事务安全，MongoDB很适合业务系统中有大量“低价值”数据的场景。

• 但是应当避免在高事务安全性的系统中使用MongoDB，除非能从架构设计上保证事务安全。

　高可用性：

• MongoDB的复副集(Master-Slave)配置非常简洁方便，此外，MongoDB可以快速响应的处理单节点故障，自动、安全的完成故障转移。

• 这些特性使得MongoDB能在一个相对不稳定（如云主机）的环境中，保持高可用性。

• 表结构不明确，且数据在不断变大

   MongoDB应用举例：

• 比如豆瓣影评中有1000部电影，每部电影有1000个影评，每个影评有1000个评论，那么查询笛卡尔积是1000*1000*1000

• 而使用MongoDB只需要存储1000条数据即可完成上述数据查询

上传解压安装MongoDB

mkdir tools # 存放安装包
mkdir training # 存放安装目录
tar -zxvf mongodb-linux-x86_64-enterprise-rhel70-3.4.10.tgz -C ~/training/
vi ~/.bash_profile

# .bash_profile

MONGODB_HOME=/root/training/mongodb-linux-x86_64-enterprise-rhel70-3.4.10
export MONGODB_HOME

PATH=$MONGODB_HOME/bin:$PATH
export PATH

source ~/.bash_profile                # 生效环境变量, 使在任何位置都可以使用MongoDB命令
安装MongoDB依赖rpm包

yum install -y net-snmp*             # 安装net-snmp : 
yum install -y cyrus*                   # 安装cyrus : 

启动MongoDB
mkdir -p /data/db
mongod --dbpath /data/db      # 指定MongoDB数据库文件存储路径（从3.2版本后，MongoDB的默认的数据引擎：wiredTiger）
mongod --dbpath /data/db1 --storageEngine=mmapv1    # 早期：内存映射，可以指定参数  --storageEngine=mmapv1
使用MongoDB Web的控制台：需要在启动MongoDB的时候，指定参数: --httpinterface
mongod --dbpath /data/db1 --storageEngine=mmapv1 --httpinterface     

测试MongoDB shell

mongo              # 进入mongo shell环境
show dbs             # 查看所有db
use mydemo              # 使用我们的数据库（如果没有这个数据库会自动创建）
db.test1.insert({id:1,"name":"Tom"})    # 在test1表中插入一条数据（如果没有这个表会自动创建）