Python实现数据库之间的数据迁移

女爷i 2022-05-30 01:17 236阅读 0赞

基于Python2.7的版本环境. Python实现的数据库跨服务器(跨库)迁移, 每以5000条一查询一提交, 代码中可以自行更改每次查询提交数目.

# -*- coding: utf-8 -*-
    
    import MySQLdb
    import time
    import warnings
    
    warnings.filterwarnings("ignore")
    
    
    class ConnectMysql(object):
        def __init__(self):
    # 这里设置分页查询, 每页查询多少数据
            self.page_size = 5000
    
        def getTable(self):
            conn = MySQLdb.connect(
                host="***.***.**.**",
                user="****",
                passwd="*************",
                db='****',
                charset='utf8'
            )
            conn_local = MySQLdb.connect(
                host="********************************",
                user="**********",
                passwd="********",
                db='*******',
                charset='utf8'
            )
            cur = conn.cursor()
            cur_local = conn_local.cursor()
            cur.execute('show tables')
            tables = cur.fetchall()
            for table in tables:
                print str(table[0]).lower()
                # 需要迁移的数据库查询表的列数
                cur.execute("SELECT COUNT(*) FROM information_schema.COLUMNS WHERE table_schema='china' AND table_name='" + table[0] + "'")
                table_col_count = cur.fetchone()
                # print table_col_count[0]
                # 需要迁移的数据库查询表的结构
                cur.execute('show create table ' + table[0])
                result = cur.fetchall()
                create_sql = result[0][1]
                # 查询需要迁移的数据库表的数据条数
                cur.execute('select count(*) from ' + table[0])
                total = cur.fetchone()
                page = total[0] / self.page_size
                page1 = total[0] % self.page_size
                if page1 != 0:
                    page = page + 1
    
                # 阿里云数据库创建表
                cur_local.execute("SELECT table_name FROM information_schema.`TABLES` WHERE table_schema='user' AND table_name='" + str(table[0]).lower() + "'")
                table_name = cur_local.fetchone()
                if table_name is None:
                    cur_local.execute(create_sql)
                for p in range(0, page):
                    while True:
                        try:
                            print '开始', table[0], '的第', p + 1, '页查询'
                            if p == 0:
                                limit_param = ' limit ' + str(p * self.page_size) + ',' + str(self.page_size)
                            else:
                                limit_param = ' limit ' + str(p * self.page_size + 1) + ',' + str(self.page_size)
                            cur.execute('select * from ' + table[0] + limit_param)
                            inserts = cur.fetchall()
                            print '查询成功'
                            param = ''
                            for i in range(0, table_col_count[0]):
                                param = param + '%s,'
                            print '开始插入'
                            cur_local.executemany('replace into ' + table[0] + ' values (' + param[0:-1] + ')', inserts)
                            print table[0], '的第', p + 1, '页, 插入完成, 还有', page - p - 1, '页, 任重而道远'
                            conn_local.commit()
                            break
                        except Exception as e:
                            print e
                            time.sleep(60)
                            cur = conn.cursor()
                            cur_local = conn_local.cursor()
                    print table[0], ' 插入完成'
                    print '\n \n ======================================================================== \n\n'
            cur_local.close()
            conn_local.close()
            cur.close()
            conn.close()
    
    
    if __name__ == '__main__':
        conn_mysql = ConnectMysql()
        conn_mysql.getTable()