MySQL 5.5 DBA工具 多进程dump 多进程load 多进程备份还原 Pytho

需要安装 python MySQL-python gzip ： yum install python MySQL-python gzip -y dump 脚本 1. 2.Created on 2012-8-20 3.mysql dump to load 4.@author: [e-mail] 5. 6.import os,time,MySQLdb,multiprocessing 7. 8.mysql_bak=/tmp/mysqlbak 9

需要安装 python MySQL-python gzip ： yum install python MySQL-python gzip -y

dump 脚本

1.'''''
2.Created on 2012-8-20
3.mysql dump to load
4.@author: [e-mail]
5.'''
6.import os,time,MySQLdb,multiprocessing 7.
8.mysql_bak='/tmp/mysqlbak' 9.mysql_base='/opt/soft/mysql_5.5.25' 10.unix_socket='/tmp/mysql.sock' 11.dump_user='root' 12.dump_pwd='123456' 13.dump_database=['test','mysql'] 14.
15.def start_process(): 16. print ('MySQLdump starting', multiprocessing.current_process().name) 17.
18.class mysqldump(object): 19. def __init__(self,conf): 20. self.conf=conf 21. self.dumpdir=mysql_bak 22.
23. def dump(self): 24. #create dir 25. self.dumpdir += '/'+str(time.strftime('%Y-%m-%d-%H-%M-%S',time.localtime(time.time()))) 26. for dir in dump_database: 27. os.system('mkdir -p '+self.dumpdir+'/'+dir) 28. os.system('mkdir -p '+self.dumpdir+'/'+dir+'/schema') 29. os.system('chmod 777 -R '+self.dumpdir) 30. #get create table 31. for dir in dump_database: 32. self.getschemainfo(dir) 33. #dump per table 34. self.getdbinfo() 35.
36. def getschemainfo(self,dbconf): 37. os.system(mysql_base+'/bin/mysqldump -d --add-drop-table -u'+dump_user+' -p'+dump_pwd+' -S'+unix_socket+' '+dbconf+' > '+self.dumpdir+'/'+dbconf+'/schema/schemainfo' ) 38. os.system('gzip '+self.dumpdir+'/'+dbconf+'/schema/schemainfo') 39. os.system(mysql_base+'/bin/mysqldump -tdRE -u'+dump_user+' -p'+dump_pwd+' -S'+unix_socket+' '+dbconf+' > '+self.dumpdir+'/'+dbconf+'/schema/objectinfo' ) 40. os.system('gzip '+self.dumpdir+'/'+dbconf+'/schema/objectinfo') 41.
42. def getdbinfo(self): 43. con=db(self.conf) 44. sql="SELECT TABLE_SCHEMA,TABLE_NAME FROM information_schema.`TABLES` WHERE TABLE_SCHEMA IN ('"+ "','".join(dump_database) +"')" 45. re = list(con.execute(sql))
46. #inputs=list() 47. pool_size = multiprocessing.cpu_count()
48. pool = multiprocessing.Pool(processes=pool_size,initializer=start_process,)
49. for tb in re: 50. #inputs.append({'dbname':tb[0],'tablename':tb[1]}) 51. pool.apply_async(self.dumplay({'dbname':tb[0],'tablename':tb[1]})) 52. #self.dumplay({'dbname':tb[0],'tablename':tb[1]}) 53. #print inputs 54. pool.close() # no more tasks 55. pool.join() # wrap up current tasks 56.
57. def dumplay(self,dbconf): 58. loadname=self.dumpdir+"/"+dbconf['dbname']+"/"+dbconf['tablename']+".sql" 59. con=db(self.conf) 60. sql="SELECT * FROM `"+dbconf['dbname']+"`.`"+dbconf['tablename']+"` INTO OUTFILE '"+loadname+"'" 61. #print sql 62. con.executeNoQuery(sql)
63. self.dogzip(loadname) 64.
65. def dogzip(self,fileconf): 66. os.system('gzip '+fileconf) 67.
68.''''' 69.
70.'''
71.class MySQLHelper(object): 72. @staticmethod
73. def getConn(conf): 74. pot = 3306 75. if(conf.has_key('port')): 76. pot=conf['port'] 77. dbname='test' 78. if(conf.has_key('db')): 79. dbname=conf['db'] 80.
81. if(conf.has_key('socket')): 82. return MySQLdb.connect(host=conf['host'],unix_socket=conf['socket'],user=conf['user'],passwd=conf['pwd'],db=dbname) 83. else: 84. return MySQLdb.connect(host=conf['host'],port=pot,user=conf['user'],passwd=conf['pwd'],db=dbname) 85.''''' 86.
87.'''
88.class db (object): 89. def __init__(self,conf): 90. self.conn=None 91. self.conn=MySQLHelper.getConn(conf) 92.
93. def execute(self,sql,mod=''): 94. if(mod=='dict'): 95. cursor=self.conn.cursor(MySQLdb.cursors.DictCursor) 96. else: 97. cursor=self.conn.cursor() 98. cursor.execute(sql)
99. set=cursor.fetchall()
100. return set 101.
102. def executeNoQuery(self,sql,param={}): 103. cursor=self.conn.cursor() 104. try: 105. if(param=={}): 106. rownum=cursor.execute(sql)
107. else: 108. rownum=cursor.executemany(sql,param)
109. self.conn.commit() 110. return rownum 111. finally: 112. cursor.close()
113.
114. def __del__(self): 115. if (self.conn!=None): 116. self.conn.close() 117.
118.if __name__ == '__main__': 119. conf={'host':'localhost','socket':unix_socket,'user':dump_user,'pwd':dump_pwd,'db':'information_schema'} 120. dump=mysqldump(conf);
121. dump.dump();
122. print "dump success"
load 脚本

1.'''''
2.Created on 2012-8-20
3.mysql dump to load
4.@author: [e-mail]
5.'''
6.import os,MySQLdb,multiprocessing 7.
8.mysql_bak='/tmp/mysqlbak/2012-09-18-21-44-34' 9.mysql_base='/opt/soft/mysql_5.5.25' 10.unix_socket='/tmp/mysql.sock' 11.dump_user='root' 12.dump_pwd='123456' 13.dump_database={'test':'test'}#dump_database={'test':'test','mysql':'mysql'} 14.
15.def start_process(): 16. print ('MySQLinput starting', multiprocessing.current_process().name) 17.#input shcema 18.#load data 19.#input object 20.class mysqlinput(object): 21. def __init__(self,conf): 22. self.conf=conf 23.
24. def input(self): 25. os.system('chmod 777 -R '+mysql_bak) 26. dirnames=os.listdir(mysql_bak)
27. for dirname in dirnames: 28. #print dirname 29. if dump_database.has_key(dirname): 30. self.inputschema(dirname) 31.
32. for dirname in dirnames: 33. if dump_database.has_key(dirname): 34. self.loadata(dirname) 35.
36. for dirname in dirnames: 37. if dump_database.has_key(dirname): 38. self.inputobject(dirname) 39.
40. def inputschema(self,dbconf): 41. print "gunzip -cd "+mysql_bak+"/"+dbconf+"/schema/schemainfo.gz | "+mysql_base+"/bin/mysql -u"+dump_user+" -p****** -S"+unix_socket 42. os.system("gunzip -cd "+mysql_bak+"/"+dbconf+"/schema/schemainfo.gz | "+mysql_base+"/bin/mysql -u"+dump_user+" -p"+dump_pwd+" -S"+unix_socket+" "+dbconf) 43.
44. def loadata(self,dbconf): 45. pool_size = multiprocessing.cpu_count()
46. pool = multiprocessing.Pool(processes=pool_size,initializer=start_process,)
47. filenames=os.listdir(mysql_bak+"/"+dbconf) 48. for filename in filenames: 49. filepath=mysql_bak+"/"+dbconf+"/"+filename 50. if os.path.isfile(filepath): 51. fileconf={'dbname':dbconf,'gzfile':filepath,'filename':filepath,'tablename':filename[0:len(filename)-4]} 52. c=fileconf['gzfile'] 53. if c[len(c)-3:len(c)]=='.gz': 54. fileconf={'dbname':dbconf,'gzfile':filepath,'filename':filepath[0:len(filepath)-3],'tablename':filename[0:len(filename)-7]} 55. pool.apply_async(self.mygunzip(fileconf)) 56.
57. pool.close() # no more tasks 58. pool.join() # wrap up current tasks 59.
60. def mygunzip(self,fileconf): 61. c=fileconf['gzfile'] 62. if c[len(c)-3:len(c)]=='.gz': 63. os.system("gunzip "+fileconf['gzfile']) 64.
65. self.loadpertable(fileconf) 66.
67. def loadpertable(self,fileconf): 68. sql="TRUNCATE \`"+fileconf['dbname']+"\`.\`"+fileconf['tablename']+"\`;LOAD DATA INFILE '"+fileconf['filename']+"' INTO TABLE \`"+fileconf['dbname']+"\`.\`"+fileconf['tablename']+"\`;" 69. print sql 70. os.system(mysql_base+"/bin/mysql -u"+dump_user+" -p"+dump_pwd+" -S"+unix_socket+" -e\""+sql+"\"") 71.
72. def inputobject(self,dbconf): 73. print "gunzip -cd "+mysql_bak+"/"+dbconf+"/schema/objectinfo.sql.gz | "+mysql_base+"/bin/mysql -u"+dump_user+" -p****** -S"+unix_socket 74. os.system("gunzip -cd "+mysql_bak+"/"+dbconf+"/schema/objectinfo.gz | "+mysql_base+"/bin/mysql -u"+dump_user+" -p"+dump_pwd+" -S"+unix_socket+" "+dbconf) 75.
76.if __name__ == '__main__': 77. conf={'host':'localhost','socket':unix_socket,'user':dump_user,'pwd':dump_pwd,'db':'information_schema'} 78. input=mysqlinput(conf);
79. input.input();
80. print "load success"