每天迁移MySQL历史数据到历史库Python脚本

每天迁移MySQL历史数据到历史库Python脚本,第1张

概述本文实例为大家分享了Python每天迁移MySQL历史数据到历史库的具体代码,供大家参考,具体内容如下

本文实例为大家分享了Python每天迁移MysqL历史数据到历史库的具体代码,供大家参考,具体内容如下

#!/usr/bin/env python # Coding:utf-8 __author__ = 'John'  import MysqLdb import sys import datetime import time  class ClassMigrate(object):   def _get_argv(self):     self.usage = """       usage():       python daily_migration.py --source=192.168.1.4:3306/db_name:tab_name/proxy/password \                     --dest=192.168.1.150:13301/db_name_archive:tab_name_201601/proxy/password \                     --delete_strategy=delete --primary_key=auto_ID --date_col=ut --time_interval=180       """     if len(sys.argv) == 1:       print self.usage       sys.exit(1)     elif sys.argv[1] == '--help' or sys.argv[1] == '-h':         print self.usage         sys.exit()     elif len(sys.argv) > 2:       for i in sys.argv[1:]:         _argv = i.split('=')         if _argv[0] == '--source':           _List = _argv[1].split('/')           self.source_host = _List[0].split(':')[0]           self.source_port = int(_List[0].split(':')[1])           self.source_db = _List[1].split(':')[0]           self.source_tab = _List[1].split(':')[1]           self.source_user = _List[2]           self.source_password = _List[3]         elif _argv[0] == '--dest':           _List = _argv[1].split('/')           self.dest_host = _List[0].split(':')[0]           self.dest_port = int(_List[0].split(':')[1])           self.dest_db = _List[1].split(':')[0]           self.dest_tab = _List[1].split(':')[1]           self.dest_user = _List[2]           self.dest_password = _List[3]         elif _argv[0] == '--delete_strategy':           self.deleteStrategy = _argv[1]           if self.deleteStrategy not in ('delete','drop'):             print (self.usage)             sys.exit(1)         elif _argv[0] == '--primary_key':           self.pk = _argv[1]         elif _argv[0] == '--date_col':           self.date_col = _argv[1]         elif _argv[0] == '--time_interval':           self.interval = _argv[1]         else:           print (self.usage)           sys.exit(1)    def __init__(self):     self._get_argv() ## --------------------------------------------------------------------     self.sourcedb_conn_str = MysqLdb.connect(host=self.source_host,port=self.source_port,user=self.source_user,passwd=self.source_password,db=self.source_db,charset='utf8')     self.sourcedb_conn_str.autocommit(True)     self.destdb_conn_str = MysqLdb.connect(host=self.dest_host,port=self.dest_port,user=self.dest_user,passwd=self.dest_password,db=self.dest_db,charset='utf8')     self.destdb_conn_str.autocommit(True) ## --------------------------------------------------------------------     self.template_tab = self.source_tab + '_template'     self.step_size = 20000 ## --------------------------------------------------------------------     self._migCompleteState = False     self._deleteCompleteState = False ## --------------------------------------------------------------------     self.source_cnt = ''     self.source_min_ID = ''     self.source_max_ID = ''     self.source_checksum = ''     self.dest_cn = '' ## --------------------------------------------------------------------     self.today = time.strftime("%Y-%m-%d")     # self.today = '2016-05-30 09:59:40' def sourcedb_query(self,sql,sql_type):     try:       cr = self.sourcedb_conn_str.cursor()       cr.execute(sql)       if sql_type == 'select':         return cr.fetchall()       elif sql_type == 'dml':         rows = self.sourcedb_conn_str.affected_rows()         return rows       else:         return True     except Exception,e:       print (str(e) + "<br>")       return False     finally:       cr.close()   def destdb_query(self,sql_type,values=''):     try:       cr = self.destdb_conn_str.cursor()       if sql_type == 'select':         cr.execute(sql)         return cr.fetchall()       elif sql_type == 'insertmany':         cr.executemany(sql,values)         rows = self.destdb_conn_str.affected_rows()         return rows       else:         cr.execute(sql)         return True     except Exception,e:       print (str(e) + "<br>")       return False     finally:       cr.close()   def create_table_from_source(self):     '''''因为tab_name表的数据需要迁移到archive引擎表,所以不适合使用这种方式。 预留作其他用途。'''     try:       sql = "show create table %s;" % self.source_tab       create_str = self.sourcedb_query(sql,'select')[0][1]       create_str = create_str.replace('CREATE table','CREATE table IF NOT EXISTS')       self.destdb_query(create_str,'ddl')       return True     except Exception,e:       print (str(e) + "<br>")       return False   def create_table_from_template(self):     try:       sql = 'CREATE table IF NOT EXISTS %s like %s;' % (self.dest_tab,self.template_tab)       state = self.destdb_query(sql,'ddl')       if state:         return True       else:         return False     except Exception,e:       print (str(e + "<br>") + "<br>")       return False   def get_min_max(self):     """ 创建目标表、并获取源表需要迁移的总条数、最小ID、最大ID """     try:       print ("\nStarting Migrate at -- %s <br>") % (datetime.datetime.Now().__str__())       sql = """select count(*),IFNulL(min(%s),-1),IFNulL(max(%s),-1) from %s where %s >= CONCAT(DATE_FORMAT(DATE_ADD('%s',INTERVAL -%s day),'%%Y-%%m-%%d'),' 00:00:00') \            and %s <= CONCAT(DATE_FORMAT(DATE_ADD('%s',' 23:59:59') """ \             % (self.pk,self.pk,self.source_tab,self.date_col,self.today,self.interval,self.interval)       q = self.sourcedb_query(sql,'select')       self.source_cnt = q[0][0]       self.source_min_ID = q[0][1]       self.source_max_ID = q[0][2]       self.source_checksum = str(self.source_cnt) + '_' + str(self.source_min_ID) + '_' + str(self.source_max_ID)       if self.source_cnt == 0 or self.source_min_ID == -1 or self.source_max_ID == -1:         print ("There is 0 record in source table been matched! <br>")         return False       else:         return True     except Exception,e:       print (str(e) + "<br>")       return False    def migrate_2_destdb(self):     try:       get_min_max_ID = self.get_min_max()       if get_min_max_ID:         k = self.source_min_ID         desc_sql = "desc %s;" % self.source_tab         # self.filed = []         cols = self.sourcedb_query(desc_sql,'select')         # for j in cols:         #   self.filed.append(j[0])         fileds = "%s," * len(cols) # 源表有多少个字段,就拼凑多少个%s,拼接到insert语句         fileds = fileds.rstrip(',')         while k <= self.source_max_ID:           sql = """select * from %s where %s >= %d and %s< %d \                and %s >= CONCAT(DATE_FORMAT(DATE_ADD('%s',' 00:00:00') \                and %s <= CONCAT(DATE_FORMAT(DATE_ADD('%s',' 23:59:59') """\              % (self.source_tab,k,k+self.step_size,self.interval)           print ("\n%s <br>") % sql           starttime = datetime.datetime.Now()           results = self.sourcedb_query(sql,'select')           insert_sql = "insert into " + self.dest_tab + " values (%s)" % fileds           rows = self.destdb_query(insert_sql,'insertmany',results)           if rows == False:             print ("Insert Failed!! <br>")           else:             print ("Inserted %s rows. <br>") % rows           endtime = datetime.datetime.Now()           timeinterval = endtime - starttime           print("Elapsed :" + str(timeinterval.seconds) + '.' + str(timeinterval.microseconds) + " seconds <br>")           k += self.step_size         print ("\nInsert complete at -- %s <br>") % (datetime.datetime.Now().__str__())         return True       else:         return False     except Exception,e:       print (str(e) + "<br>")       return False     def verify_total_cnt(self):     try:       sql = """select count(*),self.dest_tab,self.interval)       dest_result = self.destdb_query(sql,'select')       self.dest_cnt = dest_result[0][0]       dest_checksum = str(self.dest_cnt) + '_' + str(dest_result[0][1]) + '_' + str(dest_result[0][2])       print ("source_checksum: %s,dest_checksum: %s <br>") % (self.source_checksum,dest_checksum)       if self.source_cnt == dest_result[0][0] and dest_result[0][0] != 0 and self.source_checksum == dest_checksum:         self._migCompleteState = True         print ("Verify successfully !!<br>")       else:         print ("Verify Failed !!<br>")         sys.exit(77)     except Exception,e:       print (str(e) + "<br>")     def drop_daily_partition(self):     try:       if self._migCompleteState:         sql = """explain partitions select * from %s where %s >= CONCAT(DATE_FORMAT(DATE_ADD('%s',' 00:00:00')                and %s <= CONCAT(DATE_FORMAT(DATE_ADD('%s',self.interval)         partition_name = self.sourcedb_query(sql,'select')         partition_name = partition_name[0][3]             sql = """select count(*),-1) from %s partition (%s)""" \             % (self.pk,partition_name)         q = self.sourcedb_query(sql,'select')         source_cnt = q[0][0]         source_min_ID = q[0][1]         source_max_ID = q[0][2]         checksum = str(source_cnt) + '_' + str(source_min_ID) + '_' + str(source_max_ID)         if source_cnt == 0 or source_min_ID == -1 or source_max_ID == -1:           print ("There is 0 record in source PARTITION been matched! <br>")         else:           if checksum == self.source_checksum:             drop_par_sql = "alter table %s drop partition %s;" % (self.source_tab,partition_name)             droped = self.sourcedb_query(drop_par_sql,'ddl')             if droped:               print (drop_par_sql + " <br>")               print ("\nDrop partition complete at -- %s <br>") % (datetime.datetime.Now().__str__())               self._deleteCompleteState = True             else:               print (drop_par_sql + " <br>")               print ("Drop partition Failed.. <br>")           else:             print ("The partition %s checksum Failed !! Drop Failed !!") % partition_name             sys.exit(77)     except Exception,e:       print (str(e) + "<br>")    def delete_data(self):     try:       if self._migCompleteState:         k = self.source_min_ID         while k <= self.source_max_ID:           sql = """delete from %s where %s >= %d and %s< %d \                and %s >= CONCAT(DATE_FORMAT(DATE_ADD('%s',' 23:59:59') """ \              % (self.source_tab,self.interval)           print ("\n%s <br>") % sql           starttime = datetime.datetime.Now()           rows = self.sourcedb_query(sql,'dml')           if rows == False:             print ("Delete Failed!! <br>")           else:             print ("Deleted %s rows. <br>") % rows           endtime = datetime.datetime.Now()           timeinterval = endtime - starttime           print("Elapsed :" + str(timeinterval.seconds) + '.' + str(timeinterval.microseconds) + " seconds <br>")           time.sleep(1)           k += self.step_size         print ("\nDelete complete at -- %s <br>") % (datetime.datetime.Now().__str__())         self._deleteCompleteState = True     except Exception,e:       print (str(e) + "<br>")          def do(self):     tab_create = self.create_table_from_template()     if tab_create:       migration = self.migrate_2_destdb()       if migration:         self.verify_total_cnt()         if self._migCompleteState:           if self.deleteStrategy == 'drop':             self.drop_daily_partition()           else:             self.delete_data()           print ("\n<br>")           print ("====="*5 + '<br>')           print ("source_total_cnt: %s <br>") % self.source_cnt           print ("dest_total_cnt: %s <br>") % self.dest_cnt           print ("====="*5 + '<br>')           if self._deleteCompleteState:             print ("\nFinal result: Successfully !! <br>")             sys.exit(88)           else:             print ("\nFinal result: Failed !! <br>")             sys.exit(254)     else:       print ("Create table Failed ! Exiting. . .")       sys.exit(255)   f = ClassMigrate() f.do() 

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持编程小技巧。

您可能感兴趣的文章:python实现数据库跨服务器迁移使用python和Django完成博客数据库的迁移方法Python的Flask框架中使用Flask-Migrate扩展迁移数据库的教程Python中MySQL数据迁移到MongoDB脚本的方法在Python中利用Into包整洁地进行数据迁移的教程 总结

以上是内存溢出为你收集整理的每天迁移MySQL历史数据到历史库Python脚本全部内容,希望文章能够帮你解决每天迁移MySQL历史数据到历史库Python脚本所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错,欢迎将内存溢出网站推荐给程序员好友。

欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/langs/1204423.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-06-04
下一篇 2022-06-04

发表评论

登录后才能评论

评论列表(0条)

保存