ourmysql博客中提供了 《
大表删除数据的思路
》,对于大表依据主键删除的思路是必须的，删除几千万的数据还算是比较简单的，如果你的数据库中的表高达数百亿条记录 ，删除其中的几十亿，就需要考虑可用性的问题了。上述文中的 利用生成的文本方式有些不妥。
我的方法是利用存储过程,游标，先根据条件获取要删除的主键，然后依据主键删除，考虑到删除50亿条记录耗费将近7天的时间(事后得出)，必须后台执行。使用python 工具写一个脚本，可以针对多个服务器进行并行操作。
1 在各个服务器上创建存过！
delimiter //
CREATE PROCEDURE `proc_del_tab`(in com_num int , in push_time datetime )
begin
declare curid bigint ;
DECLARE rowid bigint ;
declare no_more_departments int ;
declare curs cursor for
select id
from
tab
WHERE
v3 < push_time ;
DECLARE CONTINUE HANDLER FOR NOT FOUND SET no_more_departments = 1;
SET no_more_departments=0;
set rowid = 1 ;
set autocommit = 0 ;
open curs ;
REPEAT
fetch curs into curid ;
delete from tab where id = curid ;
set rowid = rowid + 1 ;
if rowid % com_num = 0
then
commit;
end if ;
UNTIL no_more_departments
END REPEAT;
commit ;
close curs ;
end;
//
delimiter ;
2 部署python 脚本：
#!/usr/bin/env python
from MySQLdb import *
import sys
import threading
import time
import os
def now() :
#return str('2011-01-31 00:00:00')
return str( time.strftime( '%Y-%m-%d %H:%M:%S' , time.localtime() ) )
def log( strs , logs ) :
f = file( logs , 'a' , 0 )
f.write( now() + ' ' + str(strs) + '\n' )
f.close()
def delining( cur , logs ) :
sql = "SET SQL_LOG_BIN=0"
try :
cur['dsn'].execute( sql )
except Exception , e :
log( 'Set SQL_LOG_BIN OFF' + str(e) , logs )
sql = "call proc_del_tab_yang( 3000 , '%s' )" % ('2011-01-31 00:00:00')
log( 'starting process %s' % ( cur['addr'] ) , logs )
try :
cur['dsn'].execute( sql )
except Exception , e :
log( 'Execute Procedure ' + str(e) , logs )
sql = "SET SQL_LOG_BIN=1"
try :
cur['dsn'].execute( sql )
except Exception , e :
log( 'Set SQL_LOG_BIN ON' + str(e) , logs )
log( 'process %s End' % ( cur['addr'] ) , logs )
def main() :
logs = "/root/yangql/python/del_test_tab.log"
server_list=['10.250.7.110']
luser="yang"
lpasswd="yang"
con = []
for addr in server_list :
cons = None
try :
cons = connect( host = addr , user = luser , passwd = lpasswd , port = 3307 , db = 'newcloudapp' )
except Exception , e :
log( 'On Connect %s ' % ( addr ) + str(e) , logs )
continue
con.append( { 'dsn':cons , 'addr':addr } )
cur = []
for cons in con :
try :
cur.append( { 'dsn':cons['dsn'].cursor( cursorclass = cursors.DictCursor ) , 'addr':cons['addr'] } )
except Exception , e :
log( 'On Cusros %s ' % ( cons['addr'] ) + str(e) , logs )
continue
thpool = []
for curs in cur :
th = threading.Thread(target = delining ,args=( curs , logs ) )
thpool.append( th )
for th in thpool :
th.start()
for th in thpool :
threading.Thread.join( th )
while True :
if threading.activeCount() < 2 :
break
else :
time.sleep(1)
continue
for curs in cur :
try :
curs['dsn'].close()
except Exception , e :
log( 'On Close Cusros %s ' % ( curs['addr'] ) + str(e) , logs )
continue
for cons in con :
try :
cons['dsn'].close()
except Exception , e :
log( 'On Close Connect %s ' % ( str(e) ) , logs )
continue
if __name__ == '__main__' :
main()
欢迎大家提出更好的方法。。
