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

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

2022-09-09 867 发布于湖南
版权
举报
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 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程序猿-牧云
+关注
104文章1问答
目录
打赏
0
1
1
0
22
分享
相关文章
winx_19970108018
|
2月前
|
JSON 算法 API
1688商品详情API实战：Python调用全流程与数据解析技巧
本文介绍了1688电商平台的商品详情API接口，助力电商从业者高效获取商品信息。接口可返回商品基础属性、价格体系、库存状态、图片描述及商家详情等多维度数据，支持全球化语言设置。通过Python示例代码展示了如何调用该接口，帮助用户快速上手，适用于选品分析、市场研究等场景。
winx_19970108018
217 5
小白学大数据
|
3月前
|
数据采集 NoSQL 关系型数据库
Python爬虫去重策略：增量爬取与历史数据比对
Python爬虫去重策略：增量爬取与历史数据比对
小白学大数据
136 1
小白学大数据
|
2月前
|
Web App开发 数据采集 JavaScript
动态网页爬取：Python如何获取JS加载的数据？
动态网页爬取：Python如何获取JS加载的数据？
小白学大数据
413 58
互联网课堂
|
2月前
|
人工智能 数据可视化 Python
在Python中对数据点进行标签化
本文介绍了如何在Python中使用Matplotlib和Seaborn对数据点进行标签化，提升数据可视化的信息量与可读性。通过散点图示例，展示了添加数据点标签的具体方法。标签化在标识数据点、分类数据可视化及趋势分析中具有重要作用。文章强调了根据需求选择合适工具，并保持图表清晰美观的重要性。
互联网课堂
62 15
小白学大数据
|
1月前
|
数据采集 Web App开发 JavaScript
Python爬虫解析动态网页：从渲染到数据提取
Python爬虫解析动态网页：从渲染到数据提取
小白学大数据
141 0
winx_19970108018
|
2月前
|
供应链 API 开发者
1688 商品数据接口终极指南：Python 开发者如何高效获取标题 / 价格 / 销量数据（附调试工具推荐）
1688商品列表API是阿里巴巴开放平台提供的服务，允许开发者通过API获取1688平台的商品信息（标题、价格、销量等）。适用于电商选品、比价工具、供应链管理等场景。使用时需构造请求URL，携带参数（如q、start_price、end_price等），发送HTTP请求并解析返回的JSON/XML数据。示例代码展示了如何用Python调用该API获取商品列表。
winx_19970108018
130 18
站大爷
|
3月前
|
数据采集 存储 缓存
Python爬虫与代理IP：高效抓取数据的实战指南
在数据驱动的时代，网络爬虫是获取信息的重要工具。本文详解如何用Python结合代理IP抓取数据：从基础概念（爬虫原理与代理作用）到环境搭建（核心库与代理选择），再到实战步骤（单线程、多线程及Scrapy框架应用）。同时探讨反爬策略、数据处理与存储，并强调伦理与法律边界。最后分享性能优化技巧，助您高效抓取公开数据，实现技术与伦理的平衡。
站大爷
139 4
winx_19970108018
|
3月前
|
数据采集 搜索推荐 API
Python 原生爬虫教程：京东商品列表页面数据API
京东商品列表API是电商大数据分析的重要工具，支持开发者、商家和研究人员获取京东平台商品数据。通过关键词搜索、分类筛选、价格区间等条件，可返回多维度商品信息（如名称、价格、销量等），适用于市场调研与推荐系统开发。本文介绍其功能并提供Python请求示例。接口采用HTTP GET/POST方式，支持分页、排序等功能，满足多样化数据需求。
winx_19970108018
216 5
小白学大数据
|
3月前
|
数据采集 存储 JSON
用Python爬虫抓取数据并保存为JSON的完整指南
用Python爬虫抓取数据并保存为JSON的完整指南
小白学大数据
204 3
winx_19970108018
|
3月前
|
数据采集 API 数据格式
Python 原生爬虫教程：京东商品详情页面数据API
本文介绍京东商品详情API在电商领域的应用价值及功能。该API通过商品ID获取详细信息，如基本信息、价格、库存、描述和用户评价等，支持HTTP请求（GET/POST），返回JSON或XML格式数据。对于商家优化策略、开发者构建应用（如比价网站）以及消费者快速了解商品均有重要意义。研究此API有助于推动电商业务创新与发展。
winx_19970108018
140 0

热门文章

最新文章

  • 1
    Python构建MCP服务器完整教程：5步打造专属AI工具调用系统
    0
  • 2
    抖音直播间采集提取工具，直播间匿名截流获客软件，Python开发【仅供学习】
    43
  • 3
    掌握Python装饰器：轻松统计函数执行时间
    20
  • 4
    Python编程基石：整型、浮点、字符串与布尔值完全解读
    5
  • 5
    Python f-strings：让字符串格式化更简洁高效！
    17
  • 6
    Python列表推导式：简洁高效的数据处理利器
    7
  • 7
    Python中的列表推导式：简洁高效的数据处理艺术
    0
  • 8
    Python字符串格式化利器：f-strings入门指南
    11
  • 9
    Python高效字符串格式化：f-strings的魅力
    13
  • 10
    从零复现Google Veo 3：从数据预处理到视频生成的完整Python代码实现指南
    11
  • 1
    Python中的列表推导式：简洁高效的数据处理艺术
    128
  • 2
    Python字符串格式化利器：f-strings入门指南
    125
  • 3
    Python列表推导式：简洁高效的数据处理利器
    130
  • 4
    Python高效字符串格式化：f-strings的魅力
    113
  • 5
    掌握Python装饰器：轻松统计函数执行时间
    144
  • 6
    Python f-strings：让字符串格式化更简洁高效！
    144
  • 7
    Python爬取招标信息并生成可视化分析报告
    65
  • 8
    用 Python 制作简单小游戏教程：手把手教你开发猜数字游戏
    86
  • 9
    Python3.14正式支持Free Threaded版本！
    94
  • 10
    Python web Django快速入门手册全栈版，共2590字，短小精悍
    39

    • 相关课程

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

    • 相关电子书

    更多
  • Improving Python and Spark Per
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:

    • 相关实验场景

    更多
  • 用Python画圣诞树
  • Python选择及循环结构
  • 云端Python及基本操作
  • Python网络通信程序典型应用
  • Python新手入门（Anolis OS）
  • Python新手入门

    • 推荐镜像

    更多
  • python-release
    • 下一篇
    【案例实战】SpringBoot整合阿里云文件上传OSS
    目录
    AI助理

    你好，我是AI助理

    可以解答问题、推荐解决方案等