您现在的位置是：首页 > Python

当前栏目

python做mysql数据迁移_python制作mysql数据迁移脚本

Python mysql 数据

2023-03-02 11:48:48 时间

\#!/usr/bin/env python3

\#-\*- coding:utf8 -\*-

\#author：zhanbin.liu

\#!!!!!DB必须同版本

\#python3环境 pip3 install pymysql paramiko

import os

\#from pathlib import Path

import sys

import pymysql

import paramiko

\#每次只能迁移一个DB下的表，到指定DB

\#GRANT SELECT, CREATE, RELOAD, ALTER, LOCK TABLES ON \*.\* TO 'data\_migration'@'192.168.%' IDENTIFIED BY 'data\_migration@123';

tables='sqlauto\_cluster,sqlauto\_user' \#以，分割的字符串，如a,b,c

tableList = tables.split(',')

sourceIp = '192.168.1.101'

sourceDataBase = '/data/mysql/3306/data'

sourceDbName = 'inception\_web'

sourceDataDir = os.path.join(sourceDataBase,sourceDbName)

desIp = '192.168.1.102'

desDataBase = '/data/mysql/3306/data'

desDbName = 'inception\_web'

desDataDir = os.path.join(desDataBase,desDbName)

\# for table in tableList:

\# desFile = Path("%s/%s.ibd" %(desDataDir,table))

\# print(desFile)

\# if desFile.is\_file():

\# print("ok")

\# else:

\# print("no")

comUser = 'data\_migration'

comPwd = 'data\_migration@123'

comPort = 3306

client = paramiko.SSHClient()

client.set\_missing\_host\_key\_policy(paramiko.AutoAddPolicy())

def table\_judge():

print("table\_judge")

sourceTableExist = pymysql.connect(sourceIp,comUser,comPwd,sourceDbName,comPort,charset='utf8')

desTableExist = pymysql.connect(desIp,comUser,comPwd,desDbName,comPort,charset='utf8')

sourceTables = \[\]

desTables = \[\]

cursor\_source = sourceTableExist.cursor()

cursor\_des = desTableExist.cursor()

for table in tableList:

\#print(table)

cursor\_source.execute("select TABLE\_NAME from information\_schema.TABLES where TABLE\_SCHEMA='%s' and TABLE\_NAME='%s';" % (sourceDbName,table))

sourceTable\_tmp = cursor\_source.fetchall()

cursor\_des.execute("select TABLE\_NAME from information\_schema.TABLES where TABLE\_SCHEMA='%s' and TABLE\_NAME='%s';" % (desDbName,table))

desTable\_tmp = cursor\_des.fetchall()

\#print(desTable\_tmp)

if sourceTable\_tmp is ():

sourceTables.append(table)

if desTable\_tmp is not ():

desTables.append(desTable\_tmp\[0\]\[0\])

sourceTableExist.close()

desTableExist.close()

s=d=0

if sourceTables != \[\]:

print('迁移源不存在将要迁移的表：',sourceIp,sourceDbName, sourceTables,' 请检查')

s=1

if desTables != \[\]:

print('目标库存在将要迁移的表：',desIp,desDbName,desTables,' 请移除')

d=1

if s == 1 or d == 1:

sys.exit()

def data\_sync():

print('data\_sync')

db\_source = pymysql.connect(sourceIp,comUser,comPwd,sourceDbName,comPort,charset='utf8')

db\_des = pymysql.connect(desIp,comUser,comPwd,desDbName,comPort,charset='utf8')

cursor\_db\_source = db\_source.cursor()

cursor\_db\_des = db\_des.cursor()

for table in tableList:

print("正在同步表：",table)

cursor\_db\_source.execute("show create table %s;" % (table))

createTableSQL = cursor\_db\_source.fetchall()\[0\]\[1\]

print(createTableSQL)

try:

cursor\_db\_des.execute(createTableSQL)

except Exception as error:

print(error)

cursor\_db\_source.execute("flush table %s with read lock;" % (table))

cursor\_db\_des.execute("alter table %s discard tablespace;" % (table))

client.connect(sourceIp, 22, 'root')

stdin1, stdout1, stderr1 = client.exec\_command("scp %s %s:%s " % (sourceDataDir+"/"+table+".ibd", desIp, desDataDir))

stdin2, stdout2, stderr2 = client.exec\_command("scp %s %s:%s " % (sourceDataDir+"/"+table+".cfg", desIp, desDataDir))

a\_e\_1 = stderr1.readlines()

a\_e\_2 = stderr2.readlines()

if a\_e\_1 != \[\] or a\_e\_2 != \[\]:

print(a\_e\_1,a\_e\_2)

sys.exit()

client.close()

client.connect(desIp, 22, 'root')

stdin3, stdout3, stderr3 = client.exec\_command("chown -R mysql.mysql %s\*" % (desDataDir+"/"+table))

a\_e\_3 = stderr3.readlines()

if a\_e\_3 != \[\]:

print(a\_e\_1, a\_e\_2)

sys.exit()

client.close()

\#cursor\_db\_source.execute("select sleep(10);")

cursor\_db\_source.execute("unlock tables;")

cursor\_db\_des.execute("alter table %s import tablespace;" % (table))

print("同步完成")

cursor\_db\_source.close()

cursor\_db\_des.close()

def data\_checksum():

print('data\_checksum')

db\_source = pymysql.connect(sourceIp,comUser,comPwd,sourceDbName,comPort,charset='utf8')

db\_des = pymysql.connect(desIp,comUser,comPwd,desDbName,comPort,charset='utf8')

cursor\_db\_source = db\_source.cursor()

cursor\_db\_des = db\_des.cursor()

for table in tableList:

print("正在校验表：", table)

cursor\_db\_source.execute("checksum table %s;" % (table))

ck\_s = cursor\_db\_source.fetchall()\[0\]\[1\]

cursor\_db\_des.execute("checksum table %s;" % (table))

ck\_d = cursor\_db\_des.fetchall()\[0\]\[1\]

if ck\_s != ck\_d:

print("表不一致：",table)

else:

print("表一致：",table)

cursor\_db\_source.close()

cursor\_db\_des.close()

if \_\_name\_\_ == "\_\_main\_\_":

table\_judge()

data\_sync()

data\_checksum()

print('haha')

猜你喜欢

最长无重复子串
写技术博客的一些心得分享
Java 多线程（七）：线程池
Java 多线程（五）：锁（三）
Java 多线程（四）：锁（二）
Java 多线程（三）：锁（一）
Java 多线程（二）：并发编程的三大特性
线性时间非比较类排序
Java 多线程（一）：基础
合并k个已排序的链表
HDFS 高可用分布式环境搭建
合并两个有序数组
连续子数组的最大和
HDFS 分布式环境搭建
容器盛水问题
大数加法
HDFS 伪分布式环境搭建
设计LRU缓存结构
两数之和
使用单调栈来解决的一些问题

zl程序教程

当前栏目

python做mysql数据迁移_python制作mysql数据迁移脚本

相关文章