一直学一直嗨,一直嗨一直学

教你如何运用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', cursorselect 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实现不同数据库间数据同步功能的文章就介绍到这了。

Tags:, , , ,