最近接手些mysql數據庫維護，發現mysql在批量操作方面就是個渣渣啊，比起MS SQL SERVER簡直就是“不可同日而語”。
咨詢了下MySQL的高手，對於數據遷移這種問題，一種處理方式就是直接“一步到位” ，一次性將所有數據查詢插入到另外一個表，然后再刪除原表數據；另外一種處理方式就是使用pt--archiver工具來歸檔。
然並卵，“一步到位”法太刺激，pt--archiver工具用不順手，由於目前大部分的表都以自增id為主鍵，以此為此為前提自己寫個小腳本，厚臉拿出來供各位參考：
# coding: utf-8 import MySQLdb import time # common config EXEC_DETAIL_FILE = 'exec_detail.txt' DATETIME_FORMAT = '%Y-%m-%d %X' Default_MySQL_Host = '192.168.166.169' Default_MySQL_Port = 3358 Default_MySQL_User = "mysql_admin" Default_MySQL_Password = 'mysql@Admin@Pwd' Default_MySQL_Charset = "utf8" Default_MySQL_Connect_TimeOut = 120 # Transfer Config Transfer_Database_Name = "db001" Transfer_Source_Table_Name = "tb2001" Transfer_Target_Table_Name = "tb2001_his" Transfer_Condition = "dt <'2016-10-01'" Transfer_Rows_Per_Batch = 10000 Sleep_Second_Per_Batch = 0.5 def get_time_string(dt_time): """ 獲取指定格式的時間字符串 :param dt_time: 要轉換成字符串的時間 :return: 返回指定格式的字符串 """ global DATETIME_FORMAT return time.strftime(DATETIME_FORMAT, dt_time) def get_time_string(dt_time): return time.strftime("%Y-%m-%d %X", dt_time) def highlight(s): return "%s[30;2m%s%s[1m" % (chr(27), s, chr(27)) def print_warning_message(message): """ 以紅色字體顯示消息內容 :param message: 消息內容 :return: 無返回值 """ message = str(message) print(highlight('') + "%s[31;1m%s%s[0m" % (chr(27), message, chr(27))) def print_info_message(message): """ 以綠色字體輸出提醒性的消息 :param message: 消息內容 :return: 無返回值 """ message = str(message) print(highlight('') + "%s[32;2m%s%s[0m" % (chr(27), message, chr(27))) def write_file(file_path, message): """ 將傳入的message追加寫入到file_path指定的文件中 請先創建文件所在的目錄 :param file_path: 要寫入的文件路徑 :param message: 要寫入的信息 :return: """ file_handle = open(file_path, 'a') file_handle.writelines(message) # 追加一個換行以方便瀏覽 file_handle.writelines(chr(13)) file_handle.close() def get_mysql_connection(): """ 根據默認配置返回數據庫連接 :return: 數據庫連接 """ conn = MySQLdb.connect( host=Default_MySQL_Host, port=Default_MySQL_Port, user=Default_MySQL_User, passwd=Default_MySQL_Password, connect_timeout=Default_MySQL_Connect_TimeOut, charset=Default_MySQL_Charset, db=Transfer_Database_Name ) return conn def mysql_exec(sql_script, sql_param=None): """ 執行傳入的腳本，返回影響行數 :param sql_script: :param sql_param: :return: 腳本最后一條語句執行影響行數 """ try: conn = get_mysql_connection() print_info_message("在服務器{0}上執行腳本:{1}".format( conn.get_host_info(), sql_script)) cursor = conn.cursor() if sql_param is not None: cursor.execute(sql_script, sql_param) else: cursor.execute(sql_script) affect_rows = cursor.rowcount conn.commit() cursor.close() conn.close() return affect_rows except Exception as ex: cursor.close() conn.rollback() raise Exception(str(ex)) def mysql_exec_many(sql_script_list): """ 執行傳入的腳本，返回影響行數 :param sql_script_list: 要執行的腳本List,List中每個元素為sql_script, sql_param對 :return: 返回執行每個腳本影響的行數列表 """ try: conn = get_mysql_connection() exec_result_list = [] for sql_script, sql_param in sql_script_list: print_info_message("在服務器{0}上執行腳本:{1}".format( conn.get_host_info(), sql_script)) cursor = conn.cursor() if sql_param is not None: cursor.execute(sql_script, sql_param) else: cursor.execute(sql_script) affect_rows = cursor.rowcount exec_result_list.append("影響行數：{0}".format(affect_rows)) conn.commit() cursor.close() conn.close() return exec_result_list except Exception as ex: cursor.close() conn.rollback() raise Exception(str(ex)) def mysql_query(sql_script, sql_param=None): """ 執行傳入的SQL腳本，並返回查詢結果 :param sql_script: :param sql_param: :return: 返回SQL查詢結果 """ try: conn = get_mysql_connection() print_info_message("在服務器{0}上執行腳本:{1}".format( conn.get_host_info(), sql_script)) cursor = conn.cursor() if sql_param != '': cursor.execute(sql_script, sql_param) else: cursor.execute(sql_script) exec_result = cursor.fetchall() cursor.close() conn.close() return exec_result except Exception as ex: cursor.close() conn.close() raise Exception(str(ex)) def get_column_info_list(table_name): sql_script = """ DESC {0} """.format(table_name) column_info_list = [] query_result = mysql_query(sql_script=sql_script, sql_param=None) for row in query_result: column_name = row[0] column_key = row[3] column_info = column_name, column_key column_info_list.append(column_info) return column_info_list def get_id_range(): """ 按照傳入的表獲取要刪除數據最大ID、最小ID、刪除總行數 :return: 返回要刪除數據最大ID、最小ID、刪除總行數 """ global Transfer_Condition global Transfer_Rows_Per_Batch sql_script = """ SELECT MAX(ID) AS MAX_ID, MIN(ID) AS MIN_ID, COUNT(1) AS Total_Count FROM {0} WHERE {1}; """.format(Transfer_Source_Table_Name, Transfer_Condition) query_result = mysql_query(sql_script=sql_script, sql_param=None) max_id, min_id, total_count = query_result[0] # 此處有一坑，可能出現total_count不為0 但是max_id 和min_id 為None的情況 # 因此判斷max_id和min_id 是否為NULL if (max_id is None) or (min_id is None): max_id, min_id, total_count = 0, 0, 0 return max_id, min_id, total_count def check_env(): try: source_columns_info_list = get_column_info_list(Transfer_Source_Table_Name) target_columns_info_list = get_column_info_list(Transfer_Target_Table_Name) if len(source_columns_info_list) != len(target_columns_info_list): print_info("源表和目標表的列數不對,不滿足遷移條件") return False column_count = len(source_columns_info_list) id_flag = False for column_id in range(column_count): source_column_name, source_column_key = source_columns_info_list[column_id] target_column_name, target_column_key = target_columns_info_list[column_id] if source_column_name != target_column_name: print_info("源表和目標表的列名不匹配,不滿足遷移條件") return False if source_column_name.lower() == 'id' \ and source_column_key.lower() == 'pri' \ and target_column_name.lower() == 'id' \ and target_column_key.lower() == 'pri': id_flag = True if not id_flag: print_info("未找到為主鍵的id列，不滿足遷移條件") return False return True except Exception as ex: print_info("執行出現異常，異常為{0}".format(ex.message)) return False def main(): flag = check_env() if not flag: return loop_trans_data() def trans_data(current_min_id, current_max_id): global Transfer_Source_Table_Name global Transfer_Target_Table_Name global Transfer_Condition global Transfer_Rows_Per_Batch print_info_message("*" * 70) copy_data_script = """ INSERT INTO {0} SELECT * FROM {1} WHERE ID>={2} AND ID<{3} AND {4} ; """.format(Transfer_Target_Table_Name, Transfer_Source_Table_Name, current_min_id, current_max_id, Transfer_Condition) delete_data_script = """ DELETE FROM {0} WHERE ID IN ( SELECT ID FROM {1} WHERE ID>={2} AND ID<{3}) AND ID>={4} AND ID<{5}; """.format(Transfer_Source_Table_Name, Transfer_Target_Table_Name, current_min_id, current_max_id, current_min_id, current_max_id) sql_script_list = [] tem_sql_script = copy_data_script, None sql_script_list.append(tem_sql_script) tem_sql_script = delete_data_script, None sql_script_list.append(tem_sql_script) exec_result_list = mysql_exec_many(sql_script_list) print_info_message("執行結果:") for item in exec_result_list: print_info_message(item) def loop_trans_data(): max_id, min_id, total_count = get_id_range() if min_id == max_id: print_info_message("無數據需要結轉") return current_min_id = min_id global Transfer_Rows_Per_Batch while current_min_id <= max_id: current_max_id = current_min_id + Transfer_Rows_Per_Batch trans_data(current_min_id, current_max_id) current_percent = (current_max_id - min_id) * 100.0 / (max_id - min_id) left_rows = max_id - current_max_id if left_rows < 0: left_rows = 0 current_percent_str = "%.2f" % current_percent info = "當前復制進度{0}/{1},剩余{2},進度為{3}%".format(current_max_id, max_id, left_rows, current_percent_str) print_info_message(info) time.sleep(Sleep_Second_Per_Batch) current_min_id = current_max_id print_info_message("*" * 70) print_info_message("執行完成") if __name__ == '__main__': main()
按照各位場景的，需要修改數據庫連接信息：
還有需要遷移表的信息：
生成測試數據的mysql腳本：
CREATE TABLE `tb2001` ( `id` int(11) NOT NULL AUTO_INCREMENT, `c1` varchar(200) DEFAULT NULL, `dt` datetime DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8; create table tb2001_his like tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM mysql.user; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001; insert tb2001(c1,dt) select 'abc',date_add(localtime(),interval FLOOR(RAND() * 20000) hour) FROM tb2001;
最終運行結果如下：
顯示簡單粗暴，有興趣的可以在此基礎上修改！
=================================================================
