【KWDB 创作者计划】_Python语言——万数据插入测试

红目香薰

发布于 2025-04-13 07:01:47

2500

代码可运行

文章被收录于专栏：CSDNToQQCodeCSDNToQQCode

运行总次数：0

代码可运行

前言

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

前置文章：

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

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

数据库准备

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

DML：

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);

运行效果

SQL准备

基础测试语句

INSERT INTO users (username, password, email, phone, status)VALUES ('admin', '123123', '123@123.com', '15512345678', 1);
select * from users;

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

存储过程

-- 创建一个函数来插入 100 万条随机数据
CREATE OR REPLACE FUNCTION insert_million_users()
RETURNS void AS $$
DECLARE
    counter INTEGER := 0;
BEGIN
    -- 开始一个循环，插入 100 万条数据
    WHILE counter < 1000000 LOOP
        INSERT 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;
-- 调用函数
SELECT insert_million_users();
-- 查询插入的数据
SELECT * FROM users LIMIT 10;

发现不支持：

我在群里问了

换路子。

Python代码准备

这里的环境挺麻烦，我搞了好几回，终于搞定了。

环境需求：

命令：

pip install faker pip install psycopg2 pip install psycopg2-binary

代码：

#!/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 = """
        INSERT 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 insert data: {e}")
    sql = "SELECT * 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 insert data: {e}")
    cur.close()
    con.close()
    return
if __name__ == "__main__":
    main()

测试效果：

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

#!/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 = """
    INSERT 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 insert data: {e}")
    try:
        sql = "SELECT * 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()