KaiwuDBKaiwuDB

搜索
免费试用
首页 - 博客 - 创作者计划

KWDB 创作者计划 | Python语言——万数据插入测试

2025-05-18

原文链接：【KWDB 创作者计划】_Python语言——万数据插入测试_ITPUB博客

作者：红目香薰


前言

在衡量数据库性能的诸多指标中，插入数据的能力占据着举足轻重的地位。插入数据是数据库最基本也是最频繁的操作之一，它就像人体的呼吸，顺畅与否直接关系到整个系统的生命力。一个插入能力卓越的数据库，能够迅速高效地将源源不断的数据纳入其中，为后续的查询、分析等操作奠定坚实基础。相反，若数据库在插入数据时表现不佳，出现延迟、卡顿甚至崩溃等问题，那么即使它在其他方面具备一定优势，也难以满足现代业务对数据实时性和高效性的要求。


前置文章：

1、安装裸机版本kwdb： https://blog.itpub.net/70045375/viewspace-3079855/

2、使用KDC连接kwdb： https://blog.itpub.net/70045375/viewspace-3080324/


一、关系型数据库准备

这里我们先创建一个数据库，再创建对应的表，我这里留下DDL语句，方便实用。


DDL：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
CREATE DATABASE save_shop;
use save_shop;
CREATE TABLE IF NOT EXISTS users (
    user_id BIGSERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    password VARCHAR(255) NOT NULL,
    email VARCHAR(100),
    phone VARCHAR(20),
    status SMALLINT DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT chk_status CHECK (status IN (0, 1))
);
COMMENT ON TABLE users IS '用户表';
COMMENT ON COLUMN users.status IS '1:正常,0:禁用';
CREATE INDEX idx_username ON users (username);
CREATE INDEX idx_email ON users (email);
CREATE INDEX idx_phone ON users (phone);


运行效果

KWDB 创作者计划 | Python语言——万数据插入测试(图1)

二、SQL准备


1. 基础测试语句


1
2
ｉｎｓｅｒｔ INTO users (username, password, email, phone, status)VALUES ('admin', '123123', '123@123.com', '15512345678', 1);
ｓｅｌｅｃｔ * from users;

KWDB 创作者计划 | Python语言——万数据插入测试(图2)

插入是没有问题的，写一个存储过程看看。


2. 存储过程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
-- 创建一个函数来插入 100 万条随机数据
CREATE OR ｒｅｐｌａｃｅ FUNCTION insert_million_users()
RETURNS void AS $$
ｄｅｃｌａｒｅ
    counter INTEGER := 0;
BEGIN
    -- 开始一个循环，插入 100 万条数据
    WHILE counter < 1000000 LOOP
        ｉｎｓｅｒｔ INTO users (username, password, email, phone, status)
        VALUES (
            -- 生成随机用户名
            'user_' || counter,
            -- 生成随机密码，这里简单用 123456 替代，可根据需求修改
            'password_' || counter,
            -- 生成随机邮箱
            'user_' || counter || '@example.com',
            -- 生成随机电话号码
            '1' || lpad((random() * 999999999)::text, 9, '0'),
            -- 随机生成状态（0 或 1）
            floor(random() * 2)::smallint
        );
        counter := counter + 1;
    END LOOP;
    RETURN;
END;
$$ LANGUAGE plpgsql;
-- 调用函数
ｓｅｌｅｃｔ insert_million_users();
-- 查询插入的数据
ｓｅｌｅｃｔ * FROM users LIMIT 10;


发现不支持：

KWDB 创作者计划 | Python语言——万数据插入测试(图3)


并且我在官网上查询了，发现是没有存储过程这个信息的。

KWDB 创作者计划 | Python语言——万数据插入测试(图4)


不能使用存储过程我尝试着使用函数试试，但是也发现没有，所以我们就只能换路子了。

KWDB 创作者计划 | Python语言——万数据插入测试(图5)


3. Python代码准备


这里的环境挺麻烦，我搞了好几回，终于搞定了。我这里使用的是psycopg2来作的，因为3有一个包我始终没安装成功。这里单独介绍一些 psycopg2.

Psycopg 是 PostgreSQL 数据库适配器，专为 Python 编程语言而设计。Psycopg 完全遵循 Python DB API 2.0 规范，支持线程安全，允许多个线程共享同一连接，特别适合高并发和多线程的应用场景。

KWDB 支持用户通过 Psycopg 2 连接数据库，并执行创建、插入和查询作。本示例演示了如何通过 Psycopg 2 驱动连接和使用 KWDB。本示例使用的 Python 版本为 Python 3.10。


环境需求：

KWDB 创作者计划 | Python语言——万数据插入测试(图6)


命令：

pip install faker

pip 安装 psycopg2

pip 安装 psycopg2-binary


代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import psycopg2
import random
import string
from faker import Faker
# 初始化 Faker 实例，用于生成随机数据
fake = Faker()
def main():
    try:
        con = psycopg2.connect(database="save_shop", user="root", password="your pwd", host="47.93.6.70",port="26257")
        print("Connected!")
        con.set_session(autocommit=True)
        cur = con.cursor()
    except psycopg2.Error as e:
        print(f"Failed to connect to Kaiwudb: {e}")
    try:
    # 生成随机用户名
        username = ''.join(random.choices(string.ascii_lowercase, k=8))
        # 生成随机密码
        password = ''.join(random.choices(string.ascii_letters + string.digits, k=12))
        # 生成随机邮箱
        email = fake.email()
        # 生成随机电话号码
        phone = fake.phone_number()[:20]
        # 随机生成状态（0 或 1）
        status = random.randint(0, 1)
        # 插入数据的 SQL 语句
        insert_query = """
        ｉｎｓｅｒｔ INTO users (username, password, email, phone, status)
        VALUES (%s, %s, %s, %s, %s);
        """
        cur.execute(insert_query, (username, password, email, phone, status))
    except psycopg2.Error as e:
        print(f"Failed to ｉｎｓｅｒｔ data: {e}")
    sql = "ｓｅｌｅｃｔ * from save_shop.users"
    try:
        cur.execute(sql)
        rows = cur.fetchall()
        for row in rows:
            print(row)
    except psycopg2.Error as e:
        print(f"Failed to ｉｎｓｅｒｔ data: {e}")
    cur.close()
    con.close()
    return
if __name__ == "__main__":
    main()


测试效果：

KWDB 创作者计划 | Python语言——万数据插入测试(图7)

百万输入插入，这里要改一下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import psycopg2
import random
import string
from faker import Faker
import time
# 初始化 Faker 实例，用于生成随机数据
fake = Faker()
def main():
    try:
        con = psycopg2.connect(database="save_shop", user="root", password="qwe8403000",
                               host="47.93.6.70", port="26257")
        print("Connected!")
        con.set_session(autocommit=True)
        cur = con.cursor()
    except psycopg2.Error as e:
        print(f"Failed to connect to the database: {e}")
        return
    insert_query = """
    ｉｎｓｅｒｔ INTO users (username, password, email, phone, status)
    VALUES (%s, %s, %s, %s, %s);
    """
    start_time = time.time()
    try:
        for _ in range(10000):
            # 生成随机用户名
            username = ''.join(random.choices(string.ascii_lowercase, k=8))
            # 生成随机密码
            password = ''.join(random.choices(string.ascii_letters + string.digits, k=12))
            # 生成随机邮箱
            email = fake.email()
            # 生成随机电话号码
            phone = fake.phone_number()[:20]
            # 随机生成状态（0 或 1）
            status = random.randint(0, 1)
            cur.execute(insert_query, (username, password, email, phone, status))
            end_time = time.time()
            elapsed_time = end_time - start_time
            print(f"{_} 条.")
        print(f"共計：{elapsed_time:.2f} 秒.")
    except psycopg2.Error as e:
        print(f"Failed to ｉｎｓｅｒｔ data: {e}")
    try:
        sql = "ｓｅｌｅｃｔ * FROM users LIMIT 10"
        cur.execute(sql)
        rows = cur.fetchall()
        for row in rows:
            print(row)
    except psycopg2.Error as e:
        print(f"Failed to query data: {e}")
    cur.close()
    con.close()
if __name__ == "__main__":
    main()


我这里的代码大家能看到，我加了万次的循环以及时间的获取。

KWDB 创作者计划 | Python语言——万数据插入测试(图8)

慢慢插入中,这里就一个字，等了。

KWDB 创作者计划 | Python语言——万数据插入测试(图9)

我进行了多次测试。

KWDB 创作者计划 | Python语言——万数据插入测试(图10)

相对的时间消耗还是比较多的。

三、时序型数据库准备


这里建库的时候要有 TS 的声明：CREATE TS DATABASE ts_db;

KWDB 创作者计划 | Python语言——万数据插入测试(图11)


DDL语句

1
2
3
4
5
6
7
8
9
CREATE TABLE users (
  k_timestamp TIMESTAMP NOT NULL,
  user_age INT NOT NULL,
  user_height FLOAT,
  user_weight FLOAT
) TAGS (
      user_id INT NOT NULL,
      user_role VARCHAR(30) NOT NULL) 
PRIMARY TAGS (user_id);

KWDB 创作者计划 | Python语言——万数据插入测试(图12)


确认创建成功

KWDB 创作者计划 | Python语言——万数据插入测试(图13)


数据插入测试:

1
ｉｎｓｅｒｔ INTO users (k_timestamp, user_age, user_height, user_weight, user_id, user_role)VALUES ('2025-04-16T12:00:00Z', 30, 175.5, 70.0, 1, '张三');

KWDB 创作者计划 | Python语言——万数据插入测试(图14)

插入时间可以看到，比关系型数据库快乐很多了，才1.4毫秒，那么接下来我这里批量使用的Python写入一下看看。

我写的50条，再换算一下就知道1万的数据了。

1
2
3
4
5
6
7
8
9
10
 # 生成50条测试数据（时间戳间隔1分钟）
    base_time = '2024-07-01 10:00:00'
    data = [
        (            f'2024-07-01 10:{i:02d}:00',            220.0 + i*0.1,              
            3.0 + i*0.02,               
            20.5 + i*0.3,               
            123 + i                     
        ) 
        for i in range(50)
    ]


运行数据：

KWDB 创作者计划 | Python语言——万数据插入测试(图15)

能看到，使用时序数据插入的速度就快出来很多了，总量测出来是822.1955/50约等于16毫秒再乘以1万也才164.43秒，效果比关系型数据库的496.45秒节约出2倍多的时间，两者差异十分明显。


四、总结

本文围绕 KWDB 数据库插入性能展开深入测试与研究，旨在评估其在数据插入方面的表现，为开发者和数据库使用者提供实践参考。测试过程中遇到诸多环境配置和代码实现的问题，经探索优化后取得一定成果，整体可总结为以下几方面：

测试目的与重要性：数据库插入数据的能力是衡量其性能的关键指标之一，对系统的实时性和高效性至关重要。快速、稳定的数据插入是后续查询、分析等操作的基础，直接影响系统的整体生命力，因此对 KWDB 进行插入性能测试意义重大。

测试环境搭建：环境搭建过程复杂且充满挑战。Python 环境下，需安装 、 、 、 等多个库。起初尝试使用 时，遭遇诸多环境问题，如系统找不到指定文件、许可证分类提示等，最终更换为 才成功运行测试代码。此外，还需提前创建数据库和表，并准备相应的索引。在创建数据库和表时，使用了 、 CREATE TS DATABASE、 等语句，并添加了注释和索引以增强数据库的功能性和查询效率。psycopg2psycopg2-binarytzdatafakerpsycopg3psycopg2CREATE DATABASECREATE TABLE

测试方法与过程：测试过程涵盖多种方式。首先编写基础测试语句进行单条数据插入和查询验证，确保数据库基本操作正常。之后尝试编写存储过程插入 100 万条随机数据，但遇到语法不支持的问题。于是采用 Python 代码进行插入测试，通过循环生成随机用户名、密码、邮箱、电话号码和状态，并插入数据库。在万级数据插入测试中，通过记录插入时间来评估插入性能。

测试结果与分析：从测试结果来看，单条数据插入和查询操作均能正常执行。在 Python 实现的万条数据插入测试中，插入 10000 条数据耗时 496.45 秒，相对时间消耗较多，这表明在大规模数据插入场景下，KWDB 对关系型数据的作性能还有待进一步优化。再后面使用时序型数据库做插入测试，效果为 10000 条数据插入的耗时为 164.43 秒，两者差异相当的明显，看来 KWDN 对时序型数据插入的能力还是相当的可以呢。

经验与建议：本次测试积累了宝贵经验。在环境配置方面，需根据实际情况灵活选择合适的库和版本，避免因版本兼容性问题导致的错误。在代码实现上，应充分考虑数据库的特性和支持的语法，避免编写不兼容的代码。例如我们要插入时序数据类型，而非关系数据类型，建议是进一步优化数据插入算法，探索批量插入等方式提升插入效率，同时深入研究 KWDB 在高并发场景下的数据插入性能。

总体而言，通过本次对 KWDB 的万数据插入测试，深入了解了其在数据插入方面的性能表现和适用场景，为后续的数据库选型优化和应用开发提供了重要参考。


我这里还单独的下载了【PostgreSQL\17】，感觉缺啥就装啥，但是实际上还得根据实际情况理解，我最开始测试的是：psycopg3，但是3的环境问题好多，问题是：

KWDB 创作者计划 | Python语言——万数据插入测试(图16)

最后换成了psycopg2，就能运行了。坑我为大家填上了，希望能对大家有所帮助。


上一篇 : KWDB 创作者计划 | 用户基础操作与可视化工具KDC的连接(避坑必看)
下一篇 : KWDB 创作者计划 | 面向 AIoT 的分布式多模数据库 KaiwuDB适配之旅
相关推荐

体验全新的分布式多模数据库解决方案

企业版 社区版

  • 联系我们
  • 咨询电话：400-624-5688-7
  • 市场合作：marketing@kaiwudb.org.cn
  • 人才招聘：kw_joinus@inspur.com
  • 工作时间：周一至周五 9:00-17:30

KaiwuDB 是浪潮控股的数据库企业，面向工业物联网、数字能源、车联网、智慧产业等行业领域，提供稳定安全、高性能、易运维的创新数据软件与服务。

关注官方微信

友情链接：浪潮  

© 上海沄熹科技有限公司 Shanghai Yunxi Technology Co., Ltd.　   沪ICP备2023002175号-1    网站服务协议   |   隐私政策
400-624-5688-7
服务与支持
marketing@kaiwudb.org.cn