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

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

2022-09-09 765
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 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程序猿-牧云
目录
相关文章
无糖可乐嘟嘟
|
20天前
|
数据采集 数据可视化 数据挖掘
利用Python自动化处理Excel数据：从基础到进阶####
本文旨在为读者提供一个全面的指南，通过Python编程语言实现Excel数据的自动化处理。无论你是初学者还是有经验的开发者，本文都将帮助你掌握Pandas和openpyxl这两个强大的库，从而提升数据处理的效率和准确性。我们将从环境设置开始，逐步深入到数据读取、清洗、分析和可视化等各个环节，最终实现一个实际的自动化项目案例。 ####
无糖可乐嘟嘟
55 10
AIweker
|
13天前
|
关系型数据库 MySQL 数据库
Python处理数据库：MySQL与SQLite详解 | python小知识
本文详细介绍了如何使用Python操作MySQL和SQLite数据库，包括安装必要的库、连接数据库、执行增删改查等基本操作，适合初学者快速上手。
AIweker
90 15
历年考试不作弊
|
18天前
|
数据采集 分布式计算 大数据
构建高效的数据管道：使用Python进行ETL任务
在数据驱动的世界中，高效地处理和移动数据是至关重要的。本文将引导你通过一个实际的Python ETL（提取、转换、加载）项目，从概念到实现。我们将探索如何设计一个灵活且可扩展的数据管道，确保数据的准确性和完整性。无论你是数据工程师、分析师还是任何对数据处理感兴趣的人，这篇文章都将成为你工具箱中的宝贵资源。
历年考试不作弊
24 5
bruce_xiaowei
|
1月前
|
弹性计算 数据管理 数据库
从零开始构建员工管理系统：Python与SQLite3的完美结合
本文介绍如何使用Python和Tkinter构建一个图形界面的员工管理系统（EMS）。系统包括数据库设计、核心功能实现和图形用户界面创建。主要功能有查询、添加、删除员工信息及统计员工数量。通过本文，你将学会如何结合SQLite数据库进行数据管理，并使用Tkinter创建友好的用户界面。
bruce_xiaowei
61 2
从零开始构建员工管理系统：Python与SQLite3的完美结合
bruce_xiaowei
|
1月前
|
传感器 物联网 开发者
使用Python读取串行设备的温度数据
本文介绍了如何使用Python通过串行接口（如UART、RS-232或RS-485）读取温度传感器的数据。详细步骤包括硬件连接、安装`pyserial`库、配置串行端口、发送请求及解析响应等。适合嵌入式系统和物联网应用开发者参考。
bruce_xiaowei
46 3
张志凌
|
1月前
|
图形学 Python
SciPy 空间数据2
凸包（Convex Hull）是计算几何中的概念，指包含给定点集的所有凸集的交集。可以通过 `ConvexHull()` 方法创建凸包。示例代码展示了如何使用 `scipy` 库和 `matplotlib` 绘制给定点集的凸包。
张志凌
28 1
小王老师呀
|
1月前
|
JSON 数据格式 索引
Python中序列化/反序列化JSON格式的数据
【11月更文挑战第4天】本文介绍了 Python 中使用 `json` 模块进行序列化和反序列化的操作。序列化是指将 Python 对象（如字典、列表）转换为 JSON 字符串，主要使用 `json.dumps` 方法。示例包括基本的字典和列表序列化，以及自定义类的序列化。反序列化则是将 JSON 字符串转换回 Python 对象，使用 `json.loads` 方法。文中还提供了具体的代码示例，展示了如何处理不同类型的 Python 对象。
小王老师呀
26 1
爱专研的技术土狗
|
1月前
|
数据采集 Web App开发 iOS开发
如何使用 Python 语言的正则表达式进行网页数据的爬取？
使用 Python 进行网页数据爬取的步骤包括：1. 安装必要库（requests、re、bs4）；2. 发送 HTTP 请求获取网页内容；3. 使用正则表达式提取数据；4. 数据清洗和处理；5. 循环遍历多个页面。通过这些步骤，可以高效地从网页中提取所需信息。
爱专研的技术土狗
71 1
XIAZHI
|
1月前
|
数据采集 JavaScript 程序员
探索CSDN博客数据：使用Python爬虫技术
本文介绍了如何利用Python的requests和pyquery库爬取CSDN博客数据，包括环境准备、代码解析及注意事项，适合初学者学习。
XIAZHI
76 0
小白学大数据
|
1月前
|
数据采集 存储 分布式计算
超酷炫Python技术：交通数据的多维度分析
超酷炫Python技术：交通数据的多维度分析
小白学大数据
83 0

热门文章

最新文章

  • 1
    Rust 笔记：Rust 语言中使用 vector（向量）
  • 2
    Rust 大展拳脚的新兴领域：机密计算
  • 3
    2021 年 Rust 生态版图调研报告 | 星辰大海（下篇）
  • 4
    【Rust 实战】Rust 与 Wasm (2) —— 操作 Dom
  • 5
    【Rust】——函数（所有权）以及借用或引用
  • 6
    Rust每日一练(Leetday0003) 整数反转、字串转整数、回文数
  • 7
    【一起学Rust | 基础篇】rust函数与流程控制详解
  • 8
    Rust vs Go:常用语法对比(9)（2）
  • 9
    【一起学Rust | 基础篇】Rust基础——变量和数据类型
  • 10
    Rust 基础入门 —— 变量绑定与解构
  • 1
    Python 密码学实用指南（全）（2）
    40
  • 2
    【python】习题 第10周
    61
  • 3
    【python】习题第8周
    56
  • 4
    【python】习题第7周（上）
    88
  • 5
    【python】习题 6-10周（下）
    58
  • 6
    【python】习题 6-10周（中）
    66
  • 7
    Python 入门指南（七）（4）
    48
  • 8
    【python】习题 1-5周（中）
    90
  • 9
    【python】习题 1-5周（上）
    51
  • 10
    Python 入门指南（七）（2）
    60

    • 相关课程

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

    • 相关电子书

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

    • 相关实验场景

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

    • 推荐镜像

    更多
  • python-release
  • nodejs-release
  • debian-cd
    • 下一篇
    DataWorks售前咨询