你可以使用 Python 中的 Faker 库来生成虚拟的测试数据并插入到 MySQL 数据库中。下面是一个示例代码，演示如何使用 Faker 来插入 1000 万条数据到你提供的数据表中。
首先，你需要在 Python 中安装 Faker 和 mysql-connector-python 两个库，可以使用以下命令安装：
pip install Faker mysql-connector-python
然后，你可以使用以下示例代码来生成并插入数据：
from faker import Faker
import mysql.connector
from mysql.connector import errorcode
# 创建 Faker 实例
fake = Faker()
# MySQL 连接信息
config = {
'user': 'your_username',
'password': 'your_password',
'host': 'localhost',
'database': 'your_database',
}
try:
# 连接 MySQL
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
# 插入 1000 万条数据
for _ in range(10000000):
name = fake.name()
address = fake.address()
cert_code = fake.ssn()
mobile = fake.phone_number()
sex = fake.random_int(min=0, max=1)
query = ("INSERT INTO partition_test "
"(name, address, cert_code, mobile, sex) "
"VALUES (%s, %s, %s, %s, %s)")
data = (name, address, cert_code, mobile, sex)
cursor.execute(query, data)
cnx.commit()
print("插入数据完成")
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("用户名或密码错误")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("数据库不存在")
else:
print(err)
finally:
if 'cnx' in locals() and cnx.is_connected():
cursor.close()
cnx.close()
当需要一次性插入多条数据时，使用单个 INSERT 语句插入多行是更高效的做法。你可以通过构建一个包含多个元组的列表，并使用 MySQL 的 executemany() 方法来一次性插入多条数据。
以下是一个调整后的示例代码，它会一次性插入 10000 条数据：
from faker import Faker
import mysql.connector
from mysql.connector import errorcode
# 创建 Faker 实例
fake = Faker()
# MySQL 连接信息
config = {
'user': 'root',
'password': 'root',
'host': 'localhost',
'database': 'test',
'port': 8889
}
# 共计 2000万数据
BATCH_SIZE = 10000 # 每批插入的数据量
BATCH_COUNT = 2000 # 总共插入多少批次
try:
# 连接 MySQL
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
# 分批插入 BATCH_COUNT 万条数据
for _ in range(2000): # 总共插入 1000 万条数据，每次插入 1000 条，共需要循环 10000 次
data = []
for _ in range(BATCH_SIZE):
name = fake.name()
address = fake.address()
cert_code = fake.ssn()
mobile = fake.phone_number()
sex = fake.random_int(min=0, max=1)
created_at = fake.date_time_between(start_date='-3y', end_date='now')
updated_at = fake.date_time_between(start_date=created_at, end_date='now')
data.append((name, address, cert_code, mobile, sex, created_at, updated_at))
# 执行一次性插入
query = ("INSERT INTO partition_test "
"(name, address, cert_code, mobile, sex, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)")
cursor.executemany(query, data)
cnx.commit()
print(f"插入 {BATCH_SIZE} 条数据完成")
print(f"插入 {BATCH_COUNT} 万条数据完成")
except mysql.connector.Error as err:
if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("用户名或密码错误")
elif err.errno == errorcode.ER_BAD_DB_ERROR:
print("数据库不存在")
else:
print(err)
finally:
if 'cnx' in locals() and cnx.is_connected():
cursor.close()
cnx.close()
这个代码片段会生成一个包含 10000 个元组的列表 data，每个元组表示一条记录的数据。然后，使用 executemany() 方法一次性插入这些数据到数据库中，从而减少了与数据库的交互次数，提高了插入效率。请记得将 your_username、your_password 和 your_database 替换为你的 MySQL 连接信息。
附上数据表结构：
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for partion_test
-- ----------------------------
DROP TABLE IF EXISTS `partition_test`;
CREATE TABLE `partion_test` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(50) DEFAULT NULL,
`address` varchar(255) DEFAULT NULL,
`cert_code` varchar(255) DEFAULT NULL,
`mobile` varchar(255) DEFAULT NULL,
`sex` tinyint(1) DEFAULT NULL,
`created_at` timestamp NULL DEFAULT NULL,
`updated_at` timestamp NULL DEFAULT NULL,
PRIMARY KEY (`id`,`created_at`),
KEY `idx_cert_code` (`cert_code`) USING BTREE,
KEY `idx_mobile` (`mobile`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
SET FOREIGN_KEY_CHECKS = 1;
留言