Python,pymysql,create database,table,execute insert

python -m pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple

import uuid
import time
import threading
from datetime import datetime
import pymysqlTOTAL_ROWS=10**9
BATCH_SIZE=100000
THREAD_NUM=8total_inserted=0
batch_completed=0
progress_lock=threading.Lock()config={'host':'127.0.0.1','user':'root','password':'root','port':3306
}class Book:def __init__(self,id,name,isbn,author,comment,content,summary,title,topic):self.id=idself.name=nameself.isbn=isbnself.author=authorself.comment=commentself.content=contentself.summary=summaryself.title=titleself.topic=topicdef __str__(self):return f'Id:{self.id},Name:{self.name},ISBN:{self.isbn},Author:{self.author},Comment:{self.comment},Content:{self.content},Summary:{self.summary},Title:{self.title},Topic:{self.topic}'def to_dict(self):return {'name':self.name,'isbn':self.isbn,'author':self.author,'comment':self.comment,'content':self.content,'summary':self.summary,'title':self.title,'topic':self.topic}def get_mysql_conn():try:conn=pymysql.connect(**config)print(f'{datetime.now()} connect to mysql successfully!')return connexcept Exception as ex:print(f'{datetime.now()} exception:{str(ex)}')def get_mysql_conn_cur(conn):try:cur=conn.cursor()print(f'{datetime.now()} connect to mysql successfully!')return curexcept Exception as e:print(f'{datetime.now()},exception {str(e)}')def close_conn():try:conn=get_mysql_conn()if conn.open:conn.close()print(f'{datetime.now()},mysql connection closed')except Exception as ex:print(f'{datetime.now()} exception:{str(ex)}')def execute_sql(conn,cur,sql_str=""):try: if cur is None or conn is None:error_msg="cursor or connection is null!"print(f'{datetime.now()},{error_msg}')return Noneif not sql_str.strip():error_msg="sql_str is empty!"print(f'{datetime.now()},{error_msg}')return Nonecur.execute(sql_str)conn.commit()result=cur.fetchone()if result is None:result=cur.rowcountprint(f'{datetime.now()} execute {sql_str} successfully,affected rows:{result}')else:print(f'{datetime.now()} executed {sql_str} successfully,result:{result}')return resultexcept Exception as ex:if conn is not None:conn.rollback()error_msg=f'exception:{str(ex)}'        print(f'{datetime.now()},exception:{str(ex)}') return Nonedef create_db(db_name="db"):try:conn=get_mysql_conn()cur=get_mysql_conn_cur(conn)sql=f"create database if not exists {db_name}"execute_sql(conn,cur,sql)print(f'{datetime.now()}, execute sql successfully!\n{sql}')except Exception as ex:print(f'{datetime.now()},exception:{str(ex)}')def create_table(db_name="db",table_name="t1"):try:conn=get_mysql_conn()cur=get_mysql_conn_cur(conn)use_sql=f"use {db_name}"execute_sql(conn,cur,use_sql)create_table_sql=f"""create table if not exists {table_name}(Id bigint not null auto_increment primary key,name varchar(1000) not null default '',isbn varchar(1000) not null default '',author varchar(1000) not null default '', comment text,content text,summary text,title varchar(1000) not null default '',topic varchar(1000) not null default '',create_time datetime not null default current_timestamp,update_time datetime not null default current_timestamp on update current_timestamp)engine=innodb default charset=utf8mb4 collate=utf8mb4_general_ci"""
        execute_sql(conn,cur,create_table_sql)except Exception as ex:print(f'{datetime.now()},{str(ex)}')def insert_book_list_into_db(cnt=10000):conn=Nonecur=Nonetry:book_list=[]arr=range(1,cnt+1)for a in arr:book_list.append(Book(a,f'Name_{a}',f'ISBN_{a}_{uuid.uuid4().hex}',f'Author_{a}',f'Comment_{a}',f'Content_{a}',f'Summary_{a}',f'Title_{a}',f'Topic_{a}'))print(f'{datetime.now()} generated {len(book_list)} rows data')insert_params=[(bk.to_dict()['name'],bk.to_dict()['isbn'],bk.to_dict()['author'],bk.to_dict()['comment'],bk.to_dict()['content'],bk.to_dict()['summary'],bk.to_dict()['title'],bk.to_dict()['topic'])for bk in book_list]insert_sql="insert into t1(name,isbn,author,comment,content,summary,title,topic) values (%s,%s,%s,%s,%s,%s,%s,%s)"config['db']='db'conn=pymysql.connect(**config)cur=conn.cursor()cur.executemany(insert_sql,insert_params)conn.commit()print(f'{datetime.now()} insert {cur.rowcount} rows into t1 in batch')except pymysql.MySQLError as ex:if conn:conn.rollback()print(f'{datetime.now()},pymysql.MySQLError: {str(ex)}')except Exception as ex:if conn:conn.rollback()print(f'{datetime.now()} {str(ex)}')finally:if cur:cur.close()if conn and conn.open:conn.close()print(f'{datetime.now()} database connection is closed!')def insert_batch(batch_id,start_id,end_id):global total_inserted,batch_completedconn=Nonecur=Nonetry:book_list=[]for a in range(start_id,end_id+1):book_list.append(Book(id=a,name=f'Name_{a}',isbn=f'ISBN_{a}_{uuid.uuid4().hex}',author=f'Author_{a}',comment=f'Comment_{a}',content=f'Content_{a}',summary=f'Summary_{a}',title=f'Title_{a}',topic=f'Topic_{a}'))print(f'{datetime.now()},[batch {batch_id}] generated {len(book_list)} rows data')insert_params=[(bk.to_dict()['name'],bk.to_dict()['isbn'],bk.to_dict()['author'],bk.to_dict()['comment'],bk.to_dict()['content'],bk.to_dict()['summary'],bk.to_dict()['title'],bk.to_dict()['topic'],)for bk in book_list]insert_sql="insert into t1(name,isbn,author,comment,content,summary,title,topic) values (%s,%s,%s,%s,%s,%s,%s,%s)"config['db']='db'conn=pymysql.connect(**config)cur=conn.cursor()cur.executemany(insert_sql,insert_params)conn.commit()affected_rows=cur.rowcountwith progress_lock:total_inserted+=affected_rowsbatch_completed+=1print(f'{datetime.now()} [batch:{batch_id}] inserted {affected_rows} rows and totally inserted {total_inserted}')except pymysql.MySQLError as ex:if conn:conn.rollback()print(f'{datetime.now()}, [batch {batch_id}] database exception! {str(ex)},and has rolled back!')except Exception as ex:if conn:conn.rollback()print(f'{datetime.now()}, [batch {batch_id}] execute failed!{str(ex)} and has rolled back!')finally:if cur:cur.close()if conn and conn.open:conn.close()def thread_worker(batch_range):start_batch,end_batch=batch_rangefor batch_id in range(start_batch,end_batch+1):start_id=(batch_id-1)*BATCH_SIZE+1end_id=batch_id*BATCH_SIZEinsert_batch(batch_id,start_id,end_id)time.sleep(0.1)def split_batch_to_threads():total_batch=TOTAL_ROWS//BATCH_SIZEif TOTAL_ROWS%BATCH_SIZE!=0:total_batch+=1print(f'{datetime.now()},begin to insert 10 billion data into database| total batch:{total_batch} | batch_size:{BATCH_SIZE}| parallel threads:{THREAD_NUM}')batch_per_thread=total_batch//THREAD_NUMthreads=[]for i in range(THREAD_NUM):start_bacth=i*batch_per_thread+1end_batch=(i+1)*batch_per_thread if i<THREAD_NUM-1 else total_batcht=threading.Thread(target=thread_worker,args=((start_bacth,end_batch),),name=f'InsertedThread-{i+1}')threads.append(t)t.start()print(f'{datetime.now()} thread start {t.name},batch:{start_bacth} ~ {end_batch}')for t in threads:t.join()print(f'{datetime.now()} all threads finished,totally inserted {total_inserted} rows data')if __name__=="__main__":create_db()create_table()start_total=time.time()split_batch_to_threads()end_total=time.time()total_seconds=end_total-start_totalprint(f'{datetime.now()},totally inserted {TOTAL_ROWS} data,time cost:{total_seconds} seconds')

 

 

use db;
show tables;
select * from t1 order by id desc limit 1;