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

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

2022-09-09 930
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 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 SQLite
Python数据
Python rust
sqlite数据
sqlite python
Python程序猿-牧云
目录
相关文章
荔枝科研社
|
2月前
|
机器学习/深度学习 新能源 调度
电力系统短期负荷预测（Python代码+数据+详细文章讲解）
电力系统短期负荷预测（Python代码+数据+详细文章讲解）
荔枝科研社
187 1
兵临天下19970108016
|
2月前
|
缓存 API 网络架构
淘宝item_search_similar - 搜索相似的商品API接口，用python返回数据
淘宝联盟开放平台中，可通过“物料优选接口”（taobao.tbk.dg.optimus.material）实现“搜索相似商品”功能。该接口支持根据商品 ID 获取相似推荐商品，并返回商品信息、价格、优惠等数据，适用于商品推荐、比价等场景。本文提供基于 Python 的实现示例，包含接口调用、数据解析及结果展示。使用时需配置淘宝联盟的 appkey、appsecret 和 adzone_id，并注意接口调用频率限制和使用规范。
兵临天下19970108016
105 1
魔羯座liaotianfeile
|
29天前
|
存储 监控 API
Python实战：跨平台电商数据聚合系统的技术实现
本文介绍如何通过标准化API调用协议，实现淘宝、京东、拼多多等电商平台的商品数据自动化采集、清洗与存储。内容涵盖技术架构设计、Python代码示例及高阶应用（如价格监控系统），提供可直接落地的技术方案，帮助开发者解决多平台数据同步难题。
魔羯座liaotianfeile
56 0
站大爷
|
1月前
|
存储 JSON 算法
Python集合：高效处理无序唯一数据的利器
Python集合是一种高效的数据结构，具备自动去重、快速成员检测和无序性等特点，适用于数据去重、集合运算和性能优化等场景。本文通过实例详解其用法与技巧。
站大爷
87 0
技术交流18179014480
|
14天前
|
JSON API 数据安全/隐私保护
Python采集淘宝评论API接口及JSON数据返回全流程指南
Python采集淘宝评论API接口及JSON数据返回全流程指南
技术交流18179014480
943 6
计算机程序设计Y2013070224
|
16天前
|
数据采集 数据可视化 关系型数据库
基于python大数据的电影数据可视化分析系统
电影分析与可视化平台顺应电影产业数字化趋势，整合大数据处理、人工智能与Web技术，实现电影数据的采集、分析与可视化展示。平台支持票房、评分、观众行为等多维度分析，助力行业洞察与决策，同时提供互动界面，增强观众对电影文化的理解。技术上依托Python、MySQL、Flask、HTML等构建，融合数据采集与AI分析，提升电影行业的数据应用能力。
计算机程序设计Y2013070224
73 10
站大爷
|
8天前
|
存储 数据库 开发者
Python SQLite模块：轻量级数据库的实战指南
本文深入讲解Python内置sqlite3模块的实战应用，涵盖数据库连接、CRUD操作、事务管理、性能优化及高级特性，结合完整案例，助你快速掌握SQLite在小型项目中的高效使用，是Python开发者必备的轻量级数据库指南。
站大爷
88 0
站大爷
|
2月前
|
JSON 安全 API
Python处理JSON数据的最佳实践：从基础到进阶的实用指南
JSON作为数据交换通用格式，广泛应用于Web开发与API交互。本文详解Python处理JSON的10个关键实践，涵盖序列化、复杂结构处理、性能优化与安全编程，助开发者高效应对各类JSON数据挑战。
站大爷
140 1
计算机程序设计Y2013070224
|
25天前
|
数据可视化 大数据 数据挖掘
基于python大数据的招聘数据可视化分析系统
本系统基于Python开发，整合多渠道招聘数据，利用数据分析与可视化技术，助力企业高效决策。核心功能包括数据采集、智能分析、可视化展示及权限管理，提升招聘效率与人才管理水平，推动人力资源管理数字化转型。
计算机程序设计Y2013070224
69 0
兵临天下19970108016
|
3月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”：单线程 + 协程的效率反超 3 倍，这才是 Python 异步的正确打开方式
在 Python 爬虫中，多线程因 GIL 和切换开销效率低下，而协程通过用户态调度实现高并发，大幅提升爬取效率。本文详解协程原理、实战对比多线程性能，并提供最佳实践，助你掌握异步爬虫核心技术。
兵临天下19970108016
277 0

热门文章

最新文章

  • 1
    Python采集淘宝评论API接口及JSON数据返回全流程指南
  • 2
    VVIC seller_search 排行榜搜索接口深度分析及 Python 实现
  • 3
    Python脚本转EXE文件实战指南：从原理到操作全解析
  • 4
    Amazon item_review 商品评论接口深度分析及 Python 实现
  • 5
    热门小游戏源码（Python+PHP）下载-微信小程序游戏源码Unity发实战指南​
  • 6
    Python f-string：高效字符串格式化的艺术
  • 7
    如何使用LangChain的Python库结合DeepSeek进行多轮次对话？
  • 8
    Python列表推导式：简洁与高效的艺术
  • 9
    Python中的f-string：高效字符串格式化的利器
  • 10
    阿里巴巴 item_review 接口深度分析及 Python 实现
  • 1
    【Rust日报】用 Rust 从头实现一个 C 编译器
    402
  • 2
    探索Rust语言的内存安全特性
    234
  • 3
    Rust语言：为何备受开发者青睐？
    276
  • 4
    Rust代码在公司电脑监控软件中的内存安全监控
    720
  • 5
    【Rust日报】关于在其它语言中（特别是新语言中）能否直接调用Rust现有生态的研究
    93
  • 6
    使用Rust进行内存安全系统编程
    191
  • 7
    探索Rust语言在并发编程中的应用
    291
  • 8
    公司电脑监控软件的Rust编程实现与安全性提升
    480
  • 9
    Rust 和 Go：如何选择最适合你的编程语言
    157
  • 10
    使用Rust进行系统编程：安全性优势深度解析
    311

    • 相关课程

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

    • 相关电子书

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

    • 推荐镜像

    更多
  • python-release
    • 下一篇
    阿里云免费云服务器:个人用户每月750小时免费