AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Python，Rust大比拼：SQLite中插入10亿条数据

2022-09-09 906
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Python

在实际生活中，市场有这样的案例：写脚本来进行数据处理，比如说给数据库导入导出数据，这种任务一般来说最方便的方法是用python脚本，但是如果数据量比较大时候（比如上亿条）时候Python就会超级慢，看到无法忍受。在这种案例时候该怎么做呢，有一个外国老哥分享了自己的实践经历，并且对比了Python和Rust语言给SQLite插入十一条数据的情况，最后用Rust实现了在一分钟来完成任务。我们在此分享一下该实践过程，希望能对大家有所启迪，大家也可以尝试自己最拿手方法来实现该例子，并对比一下具体性能。

image.png

概述
案例中的任务是SQLite数据库插入10亿条的数据。表（user）数据结构和约束如下：

create table IF NOT EXISTS user

(

id INTEGER not null primary key,

area CHAR(6),

age INTEGER not null,

active INTEGER not null

);

随机生成数据。其中are列为六位数的区号（任何六位数字）。 age将是5、10 或15中的一个数字。Active为0或1。

实验环境硬件配置为：

MacBook Pro，2019（2.4 GHz 四核i5，8GB内存，256GB SSD硬盘，Big Sur 11.1）。

任务前提：

任务无需保持程序稳健性，如果进程崩溃并且所有数据都丢失了也没关系。可以再次运行脚本。

需要充分利用我的机器资源：100% CPU、8GB 内存和千兆字节的SSD空间。

无需使用真正的随机方法，stdlib随机方法即可。

Python原型
首先是原始版本的Python方法。Python标准库提供了一个SQLite模块，首先使用它编写了第一个版本。代码如下：

image.png

在该脚本中，通for循环中一一插入1000万条数据。执行花了将近15分钟。基于此进行优化迭代，提高性能。

SQLite中，每次插入都是原子性的并且为一个事务。每个事务都需要保证写入磁盘（涉及IO操作），因此可能会很慢。为了优化，可以尝试通过不同大小的批量插入，对比发现，100000是最佳选择。通过这个简单的更改，运行时间减少到了10分钟，优化了3分之一，但是仍然非常耗时。优化后，批量插入版本源码：

image.png

SQLite库优化
除了在代码层优化外，如果对于单纯的数据写入，对数据库本身搞的优化也是非常重要的。对于SQLite优化，可以做如下配置：

PRAGMA journal_mode = OFF;

PRAGMA synchronous = 0;

PRAGMA cache_size = 1000000;

PRAGMA locking_mode = EXCLUSIVE;

PRAGMA temp_store = MEMORY;

具体解释：

首先，journal_mode设置为OFF，将会关闭回滚日志，禁用 SQLite 的原子提交和回滚功能，这样在事务失败情况下，无法恢复，基于例子实例稳健性要求可以设置，但是严禁在生产环境中使用。

其次，关闭synchronous，SQLite这可以不再校验磁盘写入的数据可靠性。写入SQLite可能并不意味着它已刷新到磁盘。同样，严禁在生产环境中启用。

cache_size用户指定SQLite允许在内存中保留多少内存页。不要在生产中分配太高的的数值。

使用在EXCLUSIVE锁定模式，SQLite连接持有的锁永远不会被释放。

设置temp_store到MEMOR将使其表现得像一个内存数据库。

优化性能
对上面的两个脚本，添加 SQLite优化参数，然后重新运行：

def main():

con = sqlite3.connect(DB_NAME, isolation_level=None)

con.execute('PRAGMA journal_mode = OFF;')

con.execute('PRAGMA synchronous = 0;')

con.execute('PRAGMA cache_size = 1000000;') # give it a GB

con.execute('PRAGMA locking_mode = EXCLUSIVE;')

con.execute('PRAGMA temp_store = MEMORY;')

create_table(con)

faker(con, count=100_000_000)

优化后版本，原始版本，插入1亿行数据，大概花了10分钟；对比批量插入版本大概花了8.5分钟。

pypy版本
对比CPython PyPy在数据处理中可以提高性能，据说可以提高4倍以上的性能。本实验中也尝试编译PyPy解释器，运行脚本（代码无需修改）。

使用pypy解释器，批处理版本，插入1亿行数据只需2.5分钟。性能大概是Cpython的3.5倍，可见传说的4倍性能提高确实是真的，诚不我欺也！。同时，为了测试在纯循环插入中消耗的时间，在脚本中删除SQL指令并运行：

image.png

以上脚本在CPython耗时5.5分钟 。PyPy执行耗时1.5分钟（同样提高了3.5倍）。

Rust
在完成Python各种优化折腾有。又尝试了Rust版本的插入，对比也有个原始版本和批量插入版本。原始版本，也是每行插入：

use rusqlite::{params, Connection};

mod common;

fn faker(mut conn: Connection, count: i64) {

let tx = conn.transaction().unwrap();

for _ in 0..count {

let with_area = common::get_random_bool();

let age = common::get_random_age();

let is_active = common::get_random_active();

if with_area {

let area_code = common::get_random_area_code();

tx.execute(

"INSERT INTO user VALUES (NULL, ?, ?, ?)",

params![area_code, age, is_active],

)

.unwrap();

} else {

tx.execute(

"INSERT INTO user VALUES (NULL, NULL, ?, ?)",

params![age, is_active],

)

.unwrap();

}

}

tx.commit().unwrap();

}

fn main() {

let conn = Connection::open("basic.db").unwrap();

conn.execute_batch(

"PRAGMA journal_mode = OFF;

PRAGMA synchronous = 0;

PRAGMA cache_size = 1000000;

PRAGMA locking_mode = EXCLUSIVE;

PRAGMA temp_store = MEMORY;",

)

.expect("PRAGMA");

conn.execute(

"CREATE TABLE IF NOT EXISTS user (

id INTEGER not null primary key,

area CHAR(6),

age INTEGER not null,

active INTEGER not null)",

[],

)

.unwrap();

faker(conn, 100_000_000)

}

该版执行，大概用时3分钟。 然后我做了进一步的实验：

将rusqlite，换成sqlx异步运行。

use std::str::FromStr;

use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqliteSynchronous};

use sqlx::{ConnectOptions, Connection, Executor, SqliteConnection, Statement};

mod common;

async fn faker(mut conn: SqliteConnection, count: i64) -> Result<(), sqlx::Error> {

let mut tx = conn.begin().await?;

let stmt_with_area = tx

.prepare("INSERT INTO user VALUES (NULL, ?, ?, ?)")

.await?;

let stmt = tx

.prepare("INSERT INTO user VALUES (NULL, NULL, ?, ?)")

.await?;

for _ in 0..count {

let with_area = common::get_random_bool();

let age = common::get_random_age();

let is_active = common::get_random_active();

if with_area {

let area_code = common::get_random_area_code();

stmt_with_area

.query()

.bind(area_code)

.bind(age)

.bind(is_active)

.execute(&mut tx)

.await?;

} else {

stmt.query()

.bind(age)

.bind(is_active)

.execute(&mut tx)

.await?;

}

}

tx.commit().await?;

Ok(())

}

[tokio::main]

async fn main() -> Result<(), sqlx::Error> {

let mut conn = SqliteConnectOptions::from_str("basic_async.db")

.unwrap()

.create_if_missing(true)

.journal_mode(SqliteJournalMode::Off)

.synchronous(SqliteSynchronous::Off)

.connect()

.await?;

conn.execute("PRAGMA cache_size = 1000000;").await?;

conn.execute("PRAGMA locking_mode = EXCLUSIVE;").await?;

conn.execute("PRAGMA temp_store = MEMORY;").await?;

conn.execute(

"CREATE TABLE IF NOT EXISTS user (

id INTEGER not null primary key,

area CHAR(6),

age INTEGER not null,

active INTEGER not null);",

)

.await?;

faker(conn, 100_000_000).await?;

Ok(())

}

这个版本花了大约14分钟。性能反而下降下降了。比Python版本还要差（原因值得）。

对执行的原始SQL语句，切换到准备好的语句并在循环中插入行，但重用了准备好的语句。该版本只用了大约一分钟。

使用准备好的语句并将它们插入到50行的批次中，插入10亿条，耗时34.3 秒。

use rusqlite::{Connection, ToSql, Transaction};

mod common;

fn faker_wrapper(mut conn: Connection, count: i64) {

let tx = conn.transaction().unwrap();

faker(&tx, count);

tx.commit().unwrap();

}

fn faker(tx: &Transaction, count: i64) {

// that is, we will batch 50 inserts of rows at once

let min_batch_size: i64 = 50;

if count < min_batch_size {

panic!("count cant be less than min batch size");

}

// jeez, refactor this!

let mut with_area_params = " (NULL, ?, ?, ?),".repeat(min_batch_size as usize);

with_area_params.pop();

let with_area_params = with_area_params.as_str();

let mut without_area_params = " (NULL, NULL, ?, ?),".repeat(min_batch_size as usize);

without_area_params.pop();

let without_area_params = without_area_params.as_str();

let st1 = format!("INSERT INTO user VALUES {}", with_area_params);

let st2 = format!("INSERT INTO user VALUES {}", without_area_params);

let mut stmt_with_area = tx.prepare_cached(st1.as_str()).unwrap();

let mut stmt = tx.prepare_cached(st2.as_str()).unwrap();

for _ in 0..(count / min_batch_size) {

let with_area = common::get_random_bool();

let age = common::get_random_age();

let is_active = common::get_random_active();

let mut param_values: Vec<_> = Vec::new();

if with_area {

// lets prepare the batch

let mut vector = Vec::<(String, i8, i8)>::new();

for _ in 0..min_batch_size {

let area_code = common::get_random_area_code();

vector.push((area_code, age, is_active));

}

for batch in vector.iter() {

param_values.push(&batch.0 as &dyn ToSql);

param_values.push(&batch.1 as &dyn ToSql);

param_values.push(&batch.2 as &dyn ToSql);

}

stmt_with_area.execute(&*param_values).unwrap();

} else {

// lets prepare the batch

let mut vector = Vec::<(i8, i8)>::new();

for _ in 0..min_batch_size {

vector.push((age, is_active));

}

for batch in vector.iter() {

param_values.push(&batch.0 as &dyn ToSql);

param_values.push(&batch.1 as &dyn ToSql);

}

stmt.execute(&*param_values).unwrap();

}

}

}

fn main() {

let conn = Connection::open("basic_batched.db").unwrap();

conn.execute_batch(

"PRAGMA journal_mode = OFF;

PRAGMA synchronous = 0;

PRAGMA cache_size = 1000000;

PRAGMA locking_mode = EXCLUSIVE;

PRAGMA temp_store = MEMORY;",

)

.expect("PRAGMA");

conn.execute(

"CREATE TABLE IF NOT EXISTS user (

id INTEGER not null primary key,

area CHAR(6),

age INTEGER not null,

active INTEGER not null)",

[],

)

.unwrap();

faker_wrapper(conn, 100_000_000)

}

创建了一个线程版本，其中有一个从通道接收数据的写入线程和四个将数据推送到管道其他线程。

use rusqlite::{Connection, ToSql};

use std::sync::mpsc;

use std::sync::mpsc::{Receiver, Sender};

use std::thread;

mod common;

static MIN_BATCH_SIZE: i64 = 50;

enum ParamValues {

WithArea(Vec<(String, i8, i8)>),

WithoutArea(Vec<(i8, i8)>),

}

fn consumer(rx: Receiver) {

let mut conn = Connection::open("threaded_batched.db").unwrap();

conn.execute_batch(

"PRAGMA journal_mode = OFF;

PRAGMA synchronous = 0;

PRAGMA cache_size = 1000000;

PRAGMA locking_mode = EXCLUSIVE;

PRAGMA temp_store = MEMORY;",

)

.expect("PRAGMA");

conn.execute(

"CREATE TABLE IF NOT EXISTS user (

id INTEGER not null primary key,

area CHAR(6),

age INTEGER not null,

active INTEGER not null)",

[],

)

.unwrap();

let tx = conn.transaction().unwrap();

{

// jeez, refactor this!

let mut with_area_params = " (NULL, ?, ?, ?),".repeat(MIN_BATCH_SIZE as usize);

with_area_params.pop();

let with_area_params = with_area_params.as_str();

let mut without_area_params = " (NULL, NULL, ?, ?),".repeat(MIN_BATCH_SIZE as usize);

without_area_params.pop();

let without_area_params = without_area_params.as_str();

let st1 = format!("INSERT INTO user VALUES {}", with_area_params);

let st2 = format!("INSERT INTO user VALUES {}", without_area_params);

let mut stmt_with_area = tx.prepare_cached(st1.as_str()).unwrap();

let mut stmt_without_area = tx.prepare_cached(st2.as_str()).unwrap();

for param_values in rx {

let mut row_values: Vec<&dyn ToSql> = Vec::new();

match param_values {

ParamValues::WithArea(values) => {

for batch in values.iter() {

row_values.push(&batch.0 as &dyn ToSql);

row_values.push(&batch.1 as &dyn ToSql);

row_values.push(&batch.2 as &dyn ToSql);

}

stmt_with_area.execute(&*row_values).unwrap();

}

ParamValues::WithoutArea(values) => {

for batch in values.iter() {

row_values.push(&batch.0 as &dyn ToSql);

row_values.push(&batch.1 as &dyn ToSql);

}

stmt_without_area.execute(&*row_values).unwrap();

}

}

}

}

tx.commit().unwrap();

}

fn producer(tx: Sender, count: i64) {

if count < MIN_BATCH_SIZE {

panic!("count cant be less than min batch size");

}

for _ in 0..(count / MIN_BATCH_SIZE) {

let with_area = common::get_random_bool();

let age = common::get_random_age();

let is_active = common::get_random_active();

let mut param_values: Vec<_> = Vec::new();

if with_area {

// lets prepare the batch

let mut vector = Vec::<(String, i8, i8)>::new();

for _ in 0..MIN_BATCH_SIZE {

let area_code = common::get_random_area_code();

vector.push((area_code, age, is_active));

}

for batch in vector.iter() {

param_values.push(&batch.0 as &dyn ToSql);

param_values.push(&batch.1 as &dyn ToSql);

param_values.push(&batch.2 as &dyn ToSql);

}

// send the values

tx.send(ParamValues::WithArea(vector)).unwrap();

} else {

// lets prepare the batch

let mut vector = Vec::<(i8, i8)>::new();

for _ in 0..MIN_BATCH_SIZE {

vector.push((age, is_active));

}

for batch in vector.iter() {

param_values.push(&batch.0 as &dyn ToSql);

param_values.push(&batch.1 as &dyn ToSql);

}

// send the values

tx.send(ParamValues::WithoutArea(vector)).unwrap();

}

}

}

fn main() {

// setup the DB and tables

let (tx, rx): (Sender, Receiver) = mpsc::channel();

// lets launch the consumer

let consumer_handle = thread::spawn(|| consumer(rx));

let cpu_count = num_cpus::get();

let total_rows = 100_000_000;

let each_producer_count = (total_rows / cpu_count) as i64;

let mut handles = Vec::with_capacity(cpu_count);

for _ in 0..cpu_count {

let thread_tx = tx.clone();

handles.push(thread::spawn(move || {

producer(thread_tx, each_producer_count.clone())

}))

}

for t in handles {

t.join().unwrap();

}

drop(tx);

// wait till consumer is exited

consumer_handle.join().unwrap();

}

这是性能最好的版本，耗时约32.37秒。

基准测试对比并非
image.png

总结
通过案例不同任务实验，总体上可以得到：

通过SQLite PRAGMA语句优化设置可以提高插入性能。

使用准备好的语句可以提高性能

进行批量插入可以提高性能。

PyPy 实际上比CPython快4倍

线程/异步不一定能提高性能。

文章标签：
Python
测试技术
数据处理
SQL
数据库管理
固态存储
数据库
Rust
关键词：
Python数据
Python SQLite
Python rust
sqlite数据
sqlite python
Python程序猿-牧云
目录
相关文章
兵临天下19970108016
|
22天前
|
缓存 API 网络架构
淘宝item_search_similar - 搜索相似的商品API接口，用python返回数据
淘宝联盟开放平台中，可通过“物料优选接口”（taobao.tbk.dg.optimus.material）实现“搜索相似商品”功能。该接口支持根据商品 ID 获取相似推荐商品，并返回商品信息、价格、优惠等数据，适用于商品推荐、比价等场景。本文提供基于 Python 的实现示例，包含接口调用、数据解析及结果展示。使用时需配置淘宝联盟的 appkey、appsecret 和 adzone_id，并注意接口调用频率限制和使用规范。
兵临天下19970108016
75 1
小白学大数据
|
2月前
|
存储 Web App开发 前端开发
Python + Requests库爬取动态Ajax分页数据
Python + Requests库爬取动态Ajax分页数据
小白学大数据
106 1
技术交流18179014480
|
2月前
|
JSON API 数据格式
Python采集京东商品评论API接口示例，json数据返回
下面是一个使用Python采集京东商品评论的完整示例，包括API请求、JSON数据解析
技术交流18179014480
81 0
站大爷
|
24天前
|
JSON 安全 API
Python处理JSON数据的最佳实践：从基础到进阶的实用指南
JSON作为数据交换通用格式，广泛应用于Web开发与API交互。本文详解Python处理JSON的10个关键实践，涵盖序列化、复杂结构处理、性能优化与安全编程，助开发者高效应对各类JSON数据挑战。
站大爷
109 1
盹猫
|
2月前
|
XML Linux 区块链
Python提取Word表格数据教程(含.doc/.docx)
本文介绍了使用LibreOffice和python-docx库处理DOC文档表格的方法。首先需安装LibreOffice进行DOC到DOCX的格式转换，然后通过python-docx读取和修改表格数据。文中提供了详细的代码示例，包括格式转换函数、表格读取函数以及修改保存功能。该方法适用于Windows和Linux系统，解决了老旧DOC格式文档的处理难题，为需要处理历史文档的用户提供了实用解决方案。
盹猫
131 1
兵临天下19970108016
|
2月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”：单线程 + 协程的效率反超 3 倍，这才是 Python 异步的正确打开方式
在 Python 爬虫中，多线程因 GIL 和切换开销效率低下，而协程通过用户态调度实现高并发，大幅提升爬取效率。本文详解协程原理、实战对比多线程性能，并提供最佳实践，助你掌握异步爬虫核心技术。
兵临天下19970108016
223 0
winx_19970108018
|
2月前
|
JSON 数据挖掘 API
闲鱼商品列表API响应数据python解析
闲鱼商品列表API（Goodfish.item_list）提供标准化数据接口，支持GET请求，返回商品标题、价格、图片、卖家信息等。适用于电商比价、数据分析，支持多语言调用，附Python示例代码，便于开发者快速集成。
winx_19970108018
110 0
winx_19970108018
|
2月前
|
JSON 自然语言处理 API
闲鱼商品详情API响应数据python解析
闲鱼商品详情API（goodfish.item_get）通过商品ID获取标题、价格、描述、图片等信息，支持Python等多语言调用。本文提供Python请求示例，包含请求构造与数据处理方法。
winx_19970108018
124 0
winx_19970108018
|
2月前
|
JSON API 数据格式
微店商品列表API响应数据python解析
微店商品列表API为开发者提供稳定高效获取商品信息的途径，支持HTTP GET/POST请求，返回JSON格式数据，含商品ID、名称、价格、库存等字段，适用于电商数据分析与展示平台搭建等场景。本文提供Python调用示例，助您快速上手。
winx_19970108018
41 0
游客x4n62ax4r47yo
|
2月前
|
API 数据安全/隐私保护 Python
Python如何快速接入聚合数据行情API
聚合数据行情API，指的是一个接口即可提供多个不同交易品种的行情数据查询，这种接口，可以让你同时查询A股、美股、外汇等多种资产的行情数据。
游客x4n62ax4r47yo
181 0

热门文章

最新文章

  • 1
    Python 函数式编程：让代码更简洁高效
  • 2
    Python构建MCP服务器：从工具封装到AI集成的全流程实践
  • 3
    深度分析大麦网API接口，用Python脚本实现
  • 4
    Python中的异步编程：asyncio入门指南
  • 5
    requirement.txt 管理python包依赖
  • 6
    Python中的f-string：更简洁的字符串格式化
  • 7
    Python 函数式编程的执行效率：实际应用中的权衡
  • 8
    Python量化交易：结合爬虫与TA-Lib技术指标分析
  • 9
    Python中的f-string：更简洁的字符串格式化
  • 10
    Python变量与基础数据类型：整型、浮点型和字符串操作全解析
  • 1
    【Rust学习】06_切片
    94
  • 2
    员工监控软件结合 Rust 语言的可能性
    78
  • 3
    【Rust学习】05_引用和借用
    99
  • 4
    Rust与C++的区别及使用问题之对于大量使用C++实现的产品来说，迁移到Rust的问题如何解决
    163
  • 5
    Rust与C++的区别及使用问题之Rust中的bound check对性能产生影响的问题如何解决
    116
  • 6
    Rust与C++的区别及使用问题之Rust解决多线程下的共享的问题如何解决
    135
  • 7
    Rust与C++的区别及使用问题之Rust避免多线程中的lifetime的问题如何解决
    79
  • 8
    Rust与C++的区别及使用问题之Rust项目中组织目录结构的问题如何解决
    172
  • 9
    Rust与C++的区别及使用问题之Rust中函数参数传递的问题如何解决
    109
  • 10
    Rust与C++的区别及使用问题之NSA建议停止使用C++的问题如何解决
    106

    • 相关课程

    更多
  • Python爬虫实战
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 推荐镜像

    更多
  • python-release
    • 下一篇
    蛋白质语言模型 ProGen：在实验室合成由 AI 预测的蛋白质