python-rocksdb 基础使用教程

概述

python-rocksdb 是一个 Python 语言绑定的 RocksDB 键值存储库。RocksDB 是由 Facebook 开发的高性能嵌入式键值存储系统，特别适合处理大规模数据。本教程将详细介绍 python-rocksdb 的基本使用方法。

数据库初始化

基础初始化

最简单的数据库初始化方式如下：

import rocksdb

db = rocksdb.DB("test.db", rocksdb.Options(create_if_missing=True))

这行代码会在当前目录下创建一个名为 test.db 的 RocksDB 数据库，如果数据库不存在则自动创建。

生产环境配置

对于生产环境，建议进行更详细的配置：

import rocksdb

opts = rocksdb.Options()
opts.create_if_missing = True
opts.max_open_files = 300000
opts.write_buffer_size = 67108864  # 64MB
opts.max_write_buffer_number = 3
opts.target_file_size_base = 67108864  # 64MB

# 配置表工厂
opts.table_factory = rocksdb.BlockBasedTableFactory(
    filter_policy=rocksdb.BloomFilterPolicy(10),
    block_cache=rocksdb.LRUCache(2 * (1024 ** 3)),  # 2GB 缓存
    block_cache_compressed=rocksdb.LRUCache(500 * (1024 ** 2)))  # 500MB 压缩缓存

db = rocksdb.DB("test.db", opts)

这个配置包含了：

• 2.5GB 的缓存空间
• 布隆过滤器加速查找
• 64MB 内存缓冲区大小
• 更优化的文件处理参数

数据类型处理

RocksDB 将所有数据存储为未解释的字节串。在 python-rocksdb 中：

• Python 2 使用 str 类型
• Python 3 使用 bytes 类型

建议在所有字节字符串字面量前加上 b 前缀，如 b'key'，这样可以确保代码在 Python 2 和 3 中行为一致。

唯一可以传递 Unicode 对象的地方是文件系统路径，如数据库目录、WAL 目录等。

基本操作

增删改查

# 写入数据
db.put(b"key", b"value")

# 读取数据
value = db.get(b"key")

# 删除数据
db.delete(b"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")
db.put(b"key2", b"v2")

ret = db.multi_get([b"key1", b"key2", b"key3"])

print(ret[b"key1"])  # 输出 b"v1"
print(ret[b"key3"])  # 输出 None

数据迭代

基本迭代

db.put(b"key1", b"v1")
db.put(b"key2", b"v2")
db.put(b"key3", b"v3")

# 迭代键
it = db.iterkeys()
it.seek_to_first()
print(list(it))  # [b'key1', b'key2', b'key3']

# 迭代值
it = db.itervalues()
it.seek_to_first()
print(list(it))  # [b'v1', b'v2', b'v3']

# 迭代键值对
it = db.iteritems()
it.seek_to_first()
print(list(it))  # [(b'key1', b'v1'), (b'key2, b'v2'), (b'key3', b'v3')]

反向迭代

it = db.iteritems()
it.seek_to_last()
print(list(reversed(it)))  # [(b'key3', b'v3'), (b'key2', b'v2'), (b'key1', b'v1')]

SeekForPrev 操作

db.put(b'a1', b'a1_value')
db.put(b'a3', b'a3_value')
db.put(b'b1', b'b1_value')
db.put(b'b2', b'b2_value')
db.put(b'c2', b'c2_value')
db.put(b'c4', b'c4_value')

it = db.iteritems()
it.seek(b'a1')
print(it.get())  # (b'a1', b'a1_value')

it.seek_for_prev(b'c3')
print(it.get())  # (b'c2', b'c2_value')

快照

快照提供了数据库的某个时间点的只读视图：

db.put(b"a", b"1")
db.put(b"b", b"2")

snapshot = db.snapshot()
db.put(b"a", b"2")
db.delete(b"b")

# 当前视图
it = db.iteritems()
it.seek_to_first()
print(dict(it))  # {b'a': b'2'}

# 快照视图
it = db.iteritems(snapshot=snapshot)
it.seek_to_first()
print(dict(it))  # {b'a': b'1', b'b': b'2'}

合并操作符

合并操作符用于高效的读-修改-写操作。可以实现自定义的合并逻辑：

class AssocCounter(rocksdb.interfaces.AssociativeMergeOperator):
    def merge(self, key, existing_value, value):
        if existing_value:
            s = int(existing_value) + int(value)
            return (True, str(s).encode('ascii'))
        return (True, value)

    def name(self):
        return b'AssocCounter'

opts = rocksdb.Options()
opts.create_if_missing = True
opts.merge_operator = AssocCounter()
db = rocksdb.DB('test.db', opts)

db.merge(b"a", b"1")
db.merge(b"a", b"1")

print(db.get(b"a"))  # b'2'

python-rocksdb 还提供了内置的合并操作符：

from rocksdb.merge_operators import UintAddOperator, StringAppendOperator

opts = rocksdb.Options()
opts.create_if_missing = True
opts.merge_operator = UintAddOperator()
db = rocksdb.DB('/tmp/test', opts)

db.put(b'a', struct.pack('Q', 5566))
for x in range(1000):
    db.merge(b"a", struct.pack('Q', x))

print(struct.unpack('Q', db.get(b'a'))[0])  # 5566 + sum(range(1000))

前缀提取器

前缀提取器可以减少前缀范围扫描的 I/O 操作：

class StaticPrefix(rocksdb.interfaces.SliceTransform):
    def name(self):
        return b'static'

    def transform(self, src):
        return (0, 5)

    def in_domain(self, src):
        return len(src) >= 5

    def in_range(self, dst):
        return len(dst) == 5

opts = rocksdb.Options()
opts.create_if_missing=True
opts.prefix_extractor = StaticPrefix()

db = rocksdb.DB('test.db', opts)

# 填充数据
prefix = b'00002'
it = db.iteritems()
it.seek(prefix)

# 只获取指定前缀的键值对
print(dict(itertools.takewhile(lambda item: item[0].startswith(prefix), it)))

备份与恢复

创建备份

db = rocksdb.DB("test.db", rocksdb.Options(create_if_missing=True))
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")

高级配置

更改 Memtable 实现

class StaticPrefix(rocksdb.interfaces.SliceTransform):
    # ... 同上 ...

opts = rocksdb.Options()
opts.prefix_extractor = StaticPrefix()
opts.allow_concurrent_memtable_write = False
opts.memtable_factory = rocksdb.HashSkipListMemtableFactory()
opts.create_if_missing = True

db = rocksdb.DB("test.db", opts)

更改 SST 实现

opts = rocksdb.Options()
opts.table_factory = rocksdb.PlainTableFactory()
opts.create_if_missing = True

db = rocksdb.DB("test.db", opts)

更改压缩风格

opts = rocksdb.Options()
opts.compaction_style = "universal"
opts.compaction_options_universal = {"min_merge_width": 3}

db = rocksdb.DB("test.db", opts)

遍历 WriteBatch

batch = rocksdb.WriteBatch()
batch.put(b"key1", b"v1")
batch.delete(b'a')
batch.merge(b'xxx', b'value')

for op, key, value in batch:
    print(op, key, value)

输出：

Put key1 v1
Delete a
Merge xxx value

本教程涵盖了 python-rocksdb 的基本使用方法，通过这些功能可以构建高效的键值存储应用。