python实现不同数据库间数据同步功能

作者:flyingant9 时间:2024-01-18 15:58:52 

功能描述

数据库间数据同步方式很多,在上篇博文中有总结。本文是用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 < '%s' " % \
            (tablename, updatetime_s, updatetime_e)
       #print(q_sql1)
       result_list = []
       try:
         # startdatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
         cursor.execute(q_sql1)
         #results = cursor.fetchall()
         # enddatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
         # print(results) #返回是元组
         #使用流式游标迭代器+fetchone,减少内存消耗
         while True:
           result = cursor.fetchone()
           if not result:
             print("此区间无数据", q_sql1)
             break
           else:
             one_list = list(result.values())
             # print(result_list)
             result_list.append(one_list)
         print(result_list) #返回是列表
         #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()

results_len = (len(result_list))
       if results_len>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
记录每次同步任务执行的结果,或执行中发生异常的日志
此文件需要定期进行清理

来源:https://blog.csdn.net/flyingant9/article/details/113996470

标签:python,数据同步
0
投稿

猜你喜欢

  • JavaScript、tab切换完整版(自动切换、鼠标移入停止、移开运行)

    2024-04-22 22:36:28
  • Django 实现外键去除自动添加的后缀‘_id’

    2022-11-16 08:06:35
  • 利用phpexcel对数据库数据的导入excel(excel筛选)、导出excel

    2023-09-04 13:50:42
  • django之常用命令详解

    2023-02-04 07:25:19
  • Python将阿拉伯数字转换为罗马数字的方法

    2022-11-15 16:07:14
  • Iinternet Explorer浏览器简介(IE)

    2009-02-05 20:59:00
  • 参数传递解决window.open的session变量丢失

    2007-10-22 17:40:00
  • python实现0到1之间的随机数方式

    2023-08-06 17:14:50
  • javascript实现促销倒计时+fixed固定在底部

    2024-05-11 09:33:04
  • MySQL 8.0.29 安装配置方法图文教程

    2024-01-17 18:22:29
  • conn.execute(sql)后面加(0)和不加的区别

    2009-06-24 11:13:00
  • selenium+python自动化测试之页面元素定位

    2021-09-30 18:08:55
  • asp更改Windows2000管理者密码?

    2010-06-26 11:03:00
  • python标准库sys和OS的函数使用方法与实例详解

    2022-06-24 20:22:42
  • 全面理解JavaScript中的闭包

    2024-04-17 10:09:24
  • 判断字段是否被更新 新旧数据写入Audit Log表中

    2012-01-29 17:56:33
  • python实现KNN近邻算法

    2022-08-13 08:24:58
  • 什么是XSL?

    2007-11-05 19:47:00
  • python调试工具Birdseye的使用教程

    2023-08-22 19:18:48
  • 解决vue项目中某一页面不想引用公共组件app.vue的问题

    2023-07-02 16:57:35
  • asp之家 网络编程 m.aspxhome.com