文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2023-06-06 13:38

关注

这篇文章主要为大家展示了python如何实现不同数据库间数据同步功能,内容简而易懂,条理清晰,希望能够帮助大家解决疑惑,下面让小编带大家一起来研究并学习一下“python如何实现不同数据库间数据同步功能”这篇文章吧。

python是什么意思

Python是一种跨平台的、具有解释性、编译性、互动性和面向对象的脚本语言,其最初的设计是用于编写自动化脚本,随着版本的不断更新和新功能的添加,常用于用于开发独立的项目和大型项目。

功能描述

数据库间数据同步方式很多,在上篇博文中有总结。本文是用py程序实现数据同步。
A数据库中有几十张表,要汇聚到B数据库中,且表结构一致,需要准实时的进行数据同步,用工具实现时对其控制有限且配置较繁琐,故自写程序,可自由设置同步区间,记录自己想要的日志

代码

本代码实现功能简单,采用面向过程,有需求的同学可以自己优化成面向对象方式,在日志这块缺少数据监控,可根据需求增加。主要注意点:
1、数据抽取时采用区间抽取(按时间区间)、流式游标迭代器+fetchone,避免内存消耗
2、在数据插入时采用executemany(list),加快插入效率

import pymysqlimport osimport datetime,timedef 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
记录每次同步任务执行的结果,或执行中发生异常的日志
此文件需要定期进行清理

以上就是关于“python如何实现不同数据库间数据同步功能”的内容,如果改文章对你有所帮助并觉得写得不错,劳请分享给你的好友一起学习新知识,若想了解更多相关知识内容,请多多关注编程网行业资讯频道。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯