• 售前

  • 售后

热门帖子
入门百科

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

[复制链接]
塔米酉 显示全部楼层 发表于 2021-10-25 20:30:13 |阅读模式 打印 上一主题 下一主题
功能描述


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


本代码实现功能简单,接纳面向过程,有需求的同砚可以自己优化成面向对象方式,在日记这块缺少数据监控,可根据需求增加。重要留意点:
1、数据抽取时接纳区间抽取(按时间区间)、流式游标迭代器+fetchone,制止内存消耗
2、在数据插入时接纳executemany(list),加快插入服从
  1. import pymysql
  2. import os
  3. import datetime,time
  4. def update_time(content):
  5.   with open(filepathtime, 'w') as f:
  6.     f.writelines(content)
  7. def recode_log(content):
  8.   with open(filepathlog, 'a') as f:
  9.     f.writelines(content)
  10. def transferdata():
  11.   #1、获取需要抽取的表,抽取数据的时间点
  12.   with open(filepathtime, 'r') as f:
  13.     lines = f.readlines() # 读取所有数据
  14.     print("需要同步的表信息",lines)
  15.     for line in lines:
  16.       startdatetime = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
  17.       tablename_list =line.split(',')
  18.       #print(tablename_list)
  19.       #print(tablename_list[-1])
  20.       tablename_list[-1] = tablename_list[-1].replace('\n','')
  21.       #print(tablename_list)
  22.       tablename = tablename_list[0]
  23.       updatetime = tablename_list[1]
  24.       #print(tablename,updatetime)
  25.       #2、抽取此表此时间点的数据,同步
  26.       updatetime_s = datetime.datetime.strptime(updatetime, '%Y-%m-%d %H:%M:%S')
  27.       updatetime_e = (updatetime_s + datetime.timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
  28.       #print(updatetime_s)
  29.       #print(q_sql)
  30.       db = pymysql.connect(host=host_o, port=port_o, user=user_o, passwd=passwd_o, db=db_o)
  31.       cursor = db.cursor()
  32.       q_sql = "select a,b,c from %s where c >= '%s' " % \
  33.           (tablename, updatetime_s)
  34.       #2.1 首先判断下原表中是否有待同步数据,若有则同步且更新同步的时间参考点,若没有则不同步且不更新同步的时间参考点
  35.       try:
  36.         cursor.execute(q_sql)
  37.         results = cursor.fetchone()
  38.         #print(results) #返回是元组
  39.         #print("查询原表数据成功!",tablename)
  40.       except BaseException as e:
  41.         print("查询原表数据失败!",tablename, str(e))
  42.         #记录异常日志
  43.         updatetime_n = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  44.         eachline_log = updatetime_n + '[erro]:' + tablename + str(e) + '\n'
  45.         content_log.append(eachline_log)
  46.         recode_log(content_log)
  47.       db.close()
  48.       if results:
  49.         print("===============================================================================")
  50.         print("有数据可同步",tablename)
  51.         db = pymysql.connect(host=host_o, port=port_o, user=user_o, passwd=passwd_o, db=db_o, charset='utf8', cursorclass=pymysql.cursors.SSDictCursor)
  52.         cursor = db.cursor()
  53.         q_sql1 = "select a,b,c from %s where c >= '%s' and c < '%s' " % \
  54.              (tablename, updatetime_s, updatetime_e)
  55.         #print(q_sql1)
  56.         result_list = []
  57.         try:
  58.           # startdatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  59.           cursor.execute(q_sql1)
  60.           #results = cursor.fetchall()
  61.           # enddatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  62.           # print(results) #返回是元组
  63.           #使用流式游标迭代器+fetchone,减少内存消耗
  64.           while True:
  65.             result = cursor.fetchone()
  66.             if not result:
  67.               print("此区间无数据", q_sql1)
  68.               break
  69.             else:
  70.               one_list = list(result.values())
  71.               # print(result_list)
  72.               result_list.append(one_list)
  73.           print(result_list) #返回是列表
  74.           #print("查询数据成功!", tablename)
  75.         except BaseException as e:
  76.           print("查询数据失败!", tablename, str(e))
  77.           # 记录异常日志
  78.           updatetime_n = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  79.           eachline_log = updatetime_n + '[erro]:' + tablename + str(e) + '\n'
  80.           content_log.append(eachline_log)
  81.           recode_log(content_log)
  82.         db.close()
  83.         results_len = (len(result_list))
  84.         if results_len>0:
  85.           #3、将数据插入到目标表中,利用list提高插入效率
  86.           i_sql = "insert into table_t(a,b,c) values (%s,%s,%s)"
  87.           #print(i_sql)
  88.           db = pymysql.connect(host=host_d, port=port_d, user=user_d, passwd=passwd_d, db=db_d)
  89.           cursor = db.cursor()
  90.           try:
  91.             #startdatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  92.             cursor.executemany(i_sql, result_list)
  93.             db.commit()
  94.             #enddatetime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
  95.             print("插入成功!",tablename)
  96.           except BaseException as e:
  97.             db.rollback()
  98.             print("插入失败!", tablename,str(e))
  99.             #记录异常日志
  100.             updatetime_n = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  101.             eachline_log = updatetime_n + '[erro]:' + tablename + str(e) + '\n'
  102.             content_log.append(eachline_log)
  103.             recode_log(content_log)
  104.           db.close()
  105.         enddatetime = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
  106.         #4、如果有数据同步,则更新参考点时间为下一个节点时间
  107.         eachline_time = tablename+','+updatetime_e+'\n' #此时间点是下一个时间点updatetime_e
  108.         content_time.append(eachline_time)
  109.         print("更新表时间点",content_time)
  110.         # 5、记录成功日志
  111.         eachline_log = enddatetime + '[success]:' + tablename + '开始时间' + startdatetime + \
  112.           '结束时间' + enddatetime + ',同步数据量'+str(results_len)+',当前参考点' + updatetime_e + '\n'
  113.         content_log.append(eachline_log)
  114.         print("日志信息",content_log)
  115.         #print("===============================================================================")
  116.       else:
  117.         print("===============================================================================")
  118.         print("无数据可同步",tablename)
  119.         #db.close()
  120.         enddatetime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  121.         # 4、如果无数据同步,则参考点时间不更新
  122.         eachline_time = tablename + ',' + updatetime + '\n' #此时间点还是原时间updatetime
  123.         content_time.append(eachline_time)
  124.         print("不更新表时间点",content_time)
  125.         # 5、成功日志信息
  126.         eachline_log = enddatetime + '[success]:' + tablename + '开始时间' + startdatetime + \
  127.           '结束时间' + enddatetime + ',同步数据量0'+ ',当前参考点' + updatetime_e + '\n'
  128.         content_log.append(eachline_log)
  129.         print("日志信息",content_log)
  130.         #print("===============================================================================")
  131.     #更新配置文件,记录日志
  132.     update_time(content_time)
  133.     recode_log(content_log)
  134. if __name__ == '__main__':
  135.   filepathtime = 'D:/test/table-time.txt'
  136.   filepathlog = 'D:/test/table-log.txt'
  137.   host_o = 'localhost'
  138.   port_o = 3306
  139.   user_o = 'root'
  140.   passwd_o = 'root@123'
  141.   db_o = 'csdn'
  142.   host_d = 'localhost'
  143.   port_d = 3306
  144.   user_d = 'root'
  145.   passwd_d = 'root@123'
  146.   db_d = 'csdn'
  147.   content_time = []
  148.   content_log = []
  149.   transferdata()
  150.   #每5分钟执行一次同步
  151.   # while True:
  152.   #   transferdata()
  153.   #   time.sleep(300)
复制代码
table-time.txt配置文件,格式说明:
每行包罗源库表名、此表的最小时间time,以逗号分隔
若多个表,可配置多个时间
每次脚本实行后,同步更新时间time。时间间隔设置为1小时,可根据环境在updatetime_e中对增量进行修改
table-log.txt
记录每次同步使命实行的效果,或实行中发生非常的日记
此文件须要定期进行清理
到此这篇关于python实现差别数据库间数据同步功能的文章就先容到这了,更多相关python实现数据同步内容请搜索草根技术分享从前的文章或继续欣赏下面的相关文章渴望各人以后多多支持草根技术分享!

帖子地址: 

回复

使用道具 举报

分享
推广
火星云矿 | 预约S19Pro,享500抵1000!
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

草根技术分享(草根吧)是全球知名中文IT技术交流平台,创建于2021年,包含原创博客、精品问答、职业培训、技术社区、资源下载等产品服务,提供原创、优质、完整内容的专业IT技术开发社区。
  • 官方手机版

  • 微信公众号

  • 商务合作