1.打开一个数据库
import rocksdb
db = rocksdb.DB("test.db",rocksdb.Options(create_if_missing=True))
# create_if_missing=True 如果不存在则创建名为'test.db'的数据库
2.设置键值对和获取值
# RocksDB中存储的是byte strings。在python2中为str类型，在python3中为bytes类型。
db.put(b"key",b"value") #设置一个键值对，key => value
db.get(b"key") #获取键为key的值，如果不在则返回为空
b'value'
db.delete(b"key") #删除键为key为的键值对
# 将几个操作合并为一个操作
batch = rocksdb.WriteBatch()
batch.put(b"key", b"v1")
batch.delete(b"key")
batch.put(b"key", b"v2")
batch.put(b"key", b"v3")
db.write(batch)
# 一次获取多个键值对
>>> db.put(b"key1",b"v1")
>>> ret = db.multi_get([b"key",b"key1"])
>>> ret
{b'key': b'v3', b'key1': b'v1'}
3.迭代
>>> it = db.iterkeys() # 默认是无效的，先得调用seek方法
>>> it.seek_to_first()
>>> print(list(it))
[b'a', b'key', b'key1']
>>> it.seek_to_last()
>>> print(list(it))
[b'key1']
>>> it = db.itervalues() #对值进行迭代
>>> it.seek_to_first()
>>> print(list(it))
[b'b', b'v3', b'v1']
>>> it = db.iteritems() #对键值对进行迭代
>>> it.seek_to_first()
>>> print(list(it))
[(b'a', b'b'), (b'key', b'v3'), (b'key1', b'v1')]
# 反向迭代
>>> it.seek_to_last()
>>> print(list(reversed(it)))
[(b'key1', b'v1'), (b'key', b'v3'), (b'a', b'b')]
4.快照
>>> snapshot = db.snapshot()
>>> it = db.iteritems(snapshot=snapshot)
>>> it
<rocksdb._rocksdb.ItemsIterator object at 0x7f77ae179608>
>>> it.seek_to_first()
>>> print(dict(it))
{b'a': b'b', b'key': b'v3', b'key1': b'v1'}
5.备份和恢复
# 备份
backup = rocksdb.BackupEngine("test.db/backups")
backup.create_backup(db, flush_before_backup=True)
# 恢复
backup = rocksdb.BackupEngine("test.db/backups")
backup.restore_latest_backup("test.db", "test.db")