首页   

功能描述
数据库间数据同步方式很多,在上篇博文中有总结。本文是用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:
                        
                        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
记录每次同步任务执行的结果,或执行中发生异常的日志
此文件需要定期进行清理

class DBUtils(): def __init__(self): self.conn = pymysql.connect(dbDict['test1'], dbUser, dbPassword) self.cursor class SchemaMysql: #初始化用户数据 isdrop参数为是否删除目标库多余的字段 def __init__(self,src_info,des_info,isdrop): self.src_ip = sr...
1.背景: 最近公司项目的用户中心模块经常出现线上问题,测试人员需要将线上真实数据导入到测试环境中去复现Bug。公司有3套测试环境,来回切换并校验数据比较麻烦,于是就有了如下的数据库同步神器出现了。 ps:鄙人代码写的比较烂,还望各位大神多指点
通常,进程之彼此是完全孤立的,唯一的通信方式是队列或管道。但可以使用两个对象来表示共享数据。其实,这些对象使用了共享内存(通过mmap模块)使访问多个进程成为可能。Value( typecode, arg1, … argN, lock ) 在共享内容中常见ctypes对象。typecode要么是包含array模块使用的相同类型代码(如’i’,’d’等)的字符串,要么是来自ctypes模块的类型对
from pymysqlreplication import BinLogStreamReader from pymysqlreplication.row_event import DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent import sys import threading import traceback import time import pymysql mysql_sett.
最近在做python的项目,那么今天也来介绍下python的一个项目。首先先讲一下python的特点:其实python本身主要不是为了网络开发而设计的。至于现在很多web开发python用得广泛,主要还是因为python起步容易,上手快,代码易读性很高。但是值得一说的是,真正用python做项目时,其逻辑也很复杂,并不都是那么简单。这里介绍一个数据库同步python项目: # -*- co...
def __init__(self, **kwargs): self.sou_conn = kwargs['sou_conn'] self.tar_conn = kwargs['tar_conn'] """根据SQL查询数据库
python 不同数据库数据同步 最近项目有这样的需求:不同数据库中表数据同步操作。也就是从一个数据库把表数据插入到另一个数据库的表中。 import configparser import pymysql # 读取配置文件 conf = configparser.RawConfigParser() conf.read("D:\PycharmProjects\economic_relation\\venv\Include\control\conf.ini") # 获取源数据库参数 安装python第三方模块包时,报错 error: Microsoft Visual C++ 14.0 or greater is required. Get it with “Microsoft C 安装python第三方模块包时,报错 error: Microsoft Visual C++ 14.0 or greater is required. Get it with “Microsoft C 叶 落: 我直接放弃 安装python第三方模块包时,报错 error: Microsoft Visual C++ 14.0 or greater is required. Get it with “Microsoft C 小博测试成长之路: 就是为了安装个插件,要安装这么庞大的编译工具有点不太方便 python实现不同数据库间数据同步 兴趣使然的程序猿: 天下才共一石,你我一人一半 python实现不同数据库间数据同步 盼盼编程: 好文章,学习了
© 2022 微搜