共计 4551 个字符,预计需要花费 12 分钟才能阅读完成。
导读 | 这篇文章主要介绍了 python 实现不同数据库间数据同步功能, 本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下 |
功能描述
数据库间数据同步方式很多,在上篇博文中有总结。本文是用 py 程序实现数据同步。
A 数据库中有几十张表,要汇聚到 B 数据库中,且表结构一致,需要准实时的进行数据同步,用工具实现时对其控制有限且配置较繁琐,故自写程序,可自由设置同步区间,记录自己想要的日志
代码
本代码实现功能简单,采用面向过程,有需求的同学可以自己优化成面向对象方式,在日志这块缺少数据监控,可根据需求增加。主要注意点:
1、数据抽取时采用区间抽取(按时间区间)、流式游标迭代器 +fetchone,避免内存消耗
2、在数据插入时采用 executemany(list),加快插入效率
import pymysql
import os
import datetime,time
def update_time(content):
with open(filepathtime, 'w') as f:
f.writelines(content)
def recode_log(content):
with open(filepathlog, 'a') as f:
f.writelines(content)
def transferdata():
#1、获取需要抽取的表,抽取数据的时间点
with open(filepathtime, 'r') as f:
lines = f.readlines() # 读取所有数据
print("需要同步的表信息",lines)
for line in lines:
startdatetime = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
tablename_list =line.split(',')
#print(tablename_list)
#print(tablename_list[-1])
tablename_list[-1] = tablename_list[-1].replace('\n','')
#print(tablename_list)
tablename = tablename_list[0]
updatetime = tablename_list[1]
#print(tablename,updatetime)
#2、抽取此表此时间点的数据,同步
updatetime_s = datetime.datetime.strptime(updatetime, '%Y-%m-%d %H:%M:%S')
updatetime_e = (updatetime_s + datetime.timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
#print(updatetime_s)
#print(q_sql)
db = pymysql.connect(host=host_o, port=port_o, user=user_o, passwd=passwd_o, db=db_o)
cursor = db.cursor()
q_sql = "select a,b,c from %s where c >='%s' " % \
(tablename, updatetime_s)
#2.1 首先判断下原表中是否有待同步数据,若有则同步且更新同步的时间参考点,若没有则不同步且不更新同步的时间参考点
try:
cursor.execute(q_sql)
results = cursor.fetchone()
#print(results) #返回是元组
#print("查询原表数据成功!",tablename)
except BaseException as e:
print("查询原表数据失败!",tablename, str(e))
#记录异常日志
updatetime_n = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
eachline_log = updatetime_n + '[erro]:' + tablename + str(e) + '\n'
content_log.append(eachline_log)
recode_log(content_log)
db.close()
if results:
print("===============================================================================")
print("有数据可同步",tablename)
db = pymysql.connect(host=host_o, port=port_o, user=user_o, passwd=passwd_o, db=db_o, charset='utf8', cursorclass=pymysql.cursors.SSDictCursor)
cursor = db.cursor()
q_sql1 = "select a,b,c from %s where c >='%s' and c 0:
#3、将数据插入到目标表中,利用 list 提高插入效率
i_sql = "insert into table_t(a,b,c) values (%s,%s,%s)"
#print(i_sql)
db = pymysql.connect(host=host_d, port=port_d, user=user_d, passwd=passwd_d, db=db_d)
cursor = db.cursor()
try:
#startdatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
cursor.executemany(i_sql, result_list)
db.commit()
#enddatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
print("插入成功!",tablename)
except BaseException as e:
db.rollback()
print("插入失败!", tablename,str(e))
#记录异常日志
updatetime_n = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
eachline_log = updatetime_n + '[erro]:' + tablename + str(e) + '\n'
content_log.append(eachline_log)
recode_log(content_log)
db.close()
enddatetime = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
#4、如果有数据同步,则更新参考点时间为下一个节点时间
eachline_time = tablename+','+updatetime_e+'\n' #此时间点是下一个时间点 updatetime_e
content_time.append(eachline_time)
print("更新表时间点",content_time)
# 5、记录成功日志
eachline_log = enddatetime + '[success]:' + tablename + '开始时间' + startdatetime + \
'结束时间' + enddatetime + ',同步数据量'+str(results_len)+',当前参考点' + updatetime_e + '\n'
content_log.append(eachline_log)
print("日志信息",content_log)
#print("===============================================================================")
else:
print("===============================================================================")
print("无数据可同步",tablename)
#db.close()
enddatetime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
# 4、如果无数据同步,则参考点时间不更新
eachline_time = tablename + ',' + updatetime + '\n' #此时间点还是原时间 updatetime
content_time.append(eachline_time)
print("不更新表时间点",content_time)
# 5、成功日志信息
eachline_log = enddatetime + '[success]:' + tablename + '开始时间' + startdatetime + \
'结束时间' + enddatetime + ',同步数据量 0'+ ',当前参考点' + updatetime_e + '\n'
content_log.append(eachline_log)
print("日志信息",content_log)
#print("===============================================================================")
#更新配置文件,记录日志
update_time(content_time)
recode_log(content_log)
if __name__ == '__main__':
filepathtime = 'D:/test/table-time.txt'
filepathlog = 'D:/test/table-log.txt'
host_o = 'localhost'
port_o = 3306
user_o = 'root'
passwd_o = 'root@123'
db_o = 'csdn'
host_d = 'localhost'
port_d = 3306
user_d = 'root'
passwd_d = 'root@123'
db_d = 'csdn'
content_time = []
content_log = []
transferdata()
#每 5 分钟执行一次同步
# while True:
# transferdata()
# time.sleep(300)
table-time.txt 配置文件,格式说明:
每行包括源库表名、此表的最小时间 time,以逗号分隔若多个表,可配置多个时间每次脚本执行后,同步更新时间 time。时间间隔设置为 1 小时,可根据情况在 updatetime_e 中对增量进行修改
table-log.txt
记录每次同步任务执行的结果,或执行中发生异常的日志, 此文件需要定期进行清理。
到此这篇关于 python 实现不同数据库间数据同步功能的文章就介绍到这了。
正文完
星哥说事-微信公众号
![post-qrcode](https://www.xgss.net/wp-content/uploads/2023/07/二维码-微信公众号256x256.jpg)