SQLite3大数据库并发访问性能优化问题求助

一、SQLite本身的性能优化 #

1. 全内存模式加载数据库 #

既然机器能容纳300GB内存，直接把整个SQLite数据库加载到内存中，彻底消除磁盘IO开销：

• 方法一：从磁盘数据库导入到内存库

  def __init__(self):
      # 连接内存库
      self.con = sqlite3.connect(':memory:')
      # 从磁盘数据库导入数据
      disk_con = sqlite3.connect('../Data/databse.db')
      disk_con.backup(self.con)
      disk_con.close()
      # 开启内存优化设置
      self.con.execute("PRAGMA cache_size = -300000000")  # 单位KB，对应300GB缓存
      self.con.execute("PRAGMA journal_mode = OFF")  # 静态数据无需日志
      self.con.execute("PRAGMA synchronous = OFF")
  

• 方法二：设置超大缓存自动加载磁盘数据
  如果不想手动导入，可通过分配足够大的缓存，让SQLite自动将磁盘数据加载到内存：

  def __init__(self):
      self.con = sqlite3.connect('../Data/databse.db')
      self.con.execute("PRAGMA cache_size = -300000000")  # 分配300GB内存缓存
      self.con.execute("PRAGMA temp_store = MEMORY")
      self.con.execute("PRAGMA journal_mode = WAL")  # 开启WAL支持读并发
  

2. 使用参数化查询，避免动态SQL编译开销 #

当前用f-string拼接SQL会导致SQLite每次重新解析编译语句，改用参数化查询可复用查询计划，提升性能：

def query(self, value1, value2):
    cur = self.con.cursor()
    # 用?作为占位符，传入参数元组
    cur.execute(
        """
        SELECT col1, col2
        FROM test
        WHERE col1 BETWEEN ? AND ?
        """,
        (value1 - value2, value1 + value2)
    )
    rows = np.array(cur.fetchall())
    return rows

3. 正确配置并发查询 #

SQLite默认单文件模式下，多线程查询会因全局锁串行执行，反而增加线程切换开销。若要启用并发读：

• 开启WAL模式（仅适用于静态/写少读多数据集）：初始化时执行self.con.execute("PRAGMA journal_mode = WAL")，WAL模式下读操作可并行，不会被写操作阻塞（你的场景只有读，效果更明显）。
• 不要盲目提高max_concurrency：SQLite并发能力有限，建议测试max_concurrency=2或3，若无性能提升则回到1，避免不必要的线程开销。

二、Ray架构优化 #

1. 批量查询减少远程调用开销 #

当前每个worker循环1000次单独发送查询，会产生大量Ray IPC开销，改为批量提交查询：

# 修改worker函数，批量收集查询条件
@ray.remote(num_cpus=0.1)
def worker(db):
    num_values = 1000
    value1s = np.random.uniform(size=num_values)
    value2s = np.random.uniform(size=num_values)
    # 按批次打包查询条件，比如每100个一组
    batch_size = 100
    results = []
    for i in range(0, num_values, batch_size):
        batch_value1s = value1s[i:i+batch_size]
        batch_value2s = value2s[i:i+batch_size]
        # 调用批量查询方法
        batch_results = ray.get(db.batch_query.remote(batch_value1s, batch_value2s))
        for rows in batch_results:
            if len(rows):
                # 执行计算
            # 收集结果
    # 写入文件

def batch_query(self, value1s, value2s):
cur = self.con.cursor()
results = []
for v1, v2 in zip(value1s, value2s):
cur.execute(
"SELECT col1, col2 FROM test WHERE col1 BETWEEN ? AND ?",
(v1 - v2, v1 + v2)
)
results.append(np.array(cur.fetchall()))
return results

### 2. 复用游标减少资源创建开销
在`DatabaseHandler`中初始化时创建一次游标，避免每次查询重复创建：
```python
def __init__(self):
    # ... 其他初始化代码
    self.cur = self.con.cursor()  # 初始化时创建游标

def query(self, value1, value2):
    self.cur.execute(
        "SELECT col1, col2 FROM test WHERE col1 BETWEEN ? AND ?",
        (value1 - value2, value1 + value2)
    )
    rows = np.array(self.cur.fetchall())
    return rows

三、替代方案 #

1. 基于PyArrow/Parquet的内存列存方案 #

把数据集转换成Parquet列存格式，用PyArrow加载到内存，通过Ray共享内存让所有worker直接访问：

• 步骤：
  1. 一次性将原始数据转换为Parquet文件。
  2. 在Ray中创建共享的Arrow Table：

     import pyarrow.parquet as pq
     import ray
     
     @ray.remote
     def load_data():
         table = pq.read_table('../Data/dataset.parquet')
         return table
     
     shared_table = load_data.remote()
     

  3. worker直接从共享表查询：

     @ray.remote(num_cpus=0.1)
     def worker(shared_table):
         table = ray.get(shared_table)
         num_values = 1000
         value1s = np.random.uniform(size=num_values)
         value2s = np.random.uniform(size=num_values)
         for v1, v2 in zip(value1s, value2s):
             # 用PyArrow的filter做范围查询
             mask = (table['col1'] >= (v1 - v2)) & (table['col1'] <= (v1 + v2))
             filtered = table.filter(mask)
             rows = filtered.to_pandas().to_numpy()
             if len(rows):
                 # 执行计算
         # 写入文件
     

  该方案避免数据库开销，列存格式范围查询效率极高，且Ray共享内存无需复制数据。

2. 共享内存直接存储numpy数组 #

如果数据是结构化的，可将整个数据集加载到numpy数组，用Ray共享内存对象让所有worker直接访问：

import numpy as np
import ray

@ray.remote
def load_dataset():
    # 假设数据已保存为numpy格式
    data = np.load('../Data/dataset.npy')
    return data

shared_data = load_dataset.remote()

@ray.remote(num_cpus=0.1)
def worker(shared_data):
    data = ray.get(shared_data)
    num_values = 1000
    value1s = np.random.uniform(size=num_values)
    value2s = np.random.uniform(size=num_values)
    for v1, v2 in zip(value1s, value2s):
        # 直接用numpy做范围过滤
        mask = (data[:,0] >= (v1 - v2)) & (data[:,0] <= (v1 + v2))
        rows = data[mask]
        if len(rows):
            # 执行计算
    # 写入文件

该方案无中间层开销，适合简单范围查询场景。

3. Redis有序集合做范围查询 #

若需更灵活的并发查询，可将数据导入Redis的ZSET结构，用ZRANGEBYSCORE快速查询范围：

• 导入数据时，把col1作为score、col2作为value存入ZSET。
• worker直接连接Redis执行查询，无需中间DB进程，减少IPC开销。

内容的提问来源于stack exchange，提问作者DBStuck