Mysql指定部分数据同步
2024-08-28 08:45:25
一、需求背景
朋友的公司需要每天定时从源端定时同步一部分数据到目标端，库中存在company_id列的表，只将指定的company_id列导入到目标端数据库；存在company_id列的表，将表中所有的数据导入到目标端。
二、实现思路
1 远端与目标端的ip地址、账号、密码、端口号等信息保存在配置文件中，由我朋友自己填写；
2 通过information_schema.`COLUMNS`获取需要导出的表数据；
3 通过mysqldump导出表
4 通过通过mysql -e 的方式导入数据
5 预先在目标端建立相对应的schema
三、代码实现
3.1 配置文件db.conf
[db]
source_host=源端IP
source_port=源端端口
source_user=源端用户名
source_passwd=源端密码
dest_host=目标端用户名
dest_port=目标端端口
dest_user=目标端用户名
dest_passwd=目标端密码
3.2 实现代码
# -*- coding: utf-8 -*-
# @Time : 2019-09-11 10:25
# @Author : Huangwenjun
# @Email : 350920551@qq.com
# @File : rsyncdata.py
# @Software: PyCharm
# @exe_location:
import pymysql
import configparser
import os
class RsyncData():
def __init__(self):
self.get_item()
self.mysql_host = self.item['source_host']
self.mysql_port = int(self.item['source_port'])
self.mysql_user = self.item['source_user']
self.mysql_pwd = self.item['source_passwd']
self.dest_host=self.item['dest_host']
self.dest_port=int(self.item['dest_port'])
self.dest_user = self.item['dest_user']
self.dest_pwd = self.item['dest_passwd']
self.conn_source() def get_item(self):
"""
读取配置文件信息
"""
cf = configparser.ConfigParser()
cf.read('db.conf', encoding='utf8')
config = {}
for db in cf.sections():
items = {}
for item in cf.items(db):
items[item[0]] = item[1]
config[db] = items
self.item = config['db'] def conn_source(self):
"""
连接源端数据库
"""
self.source_conn = pymysql.connect(host=self.dest_host, port=self.dest_port, user=self.dest_user,
passwd=self.dest_pwd, charset='utf8mb4') def execute_sql(self, conn, sql):
"""
执行sql
"""
curs = conn.cursor()
curs.execute(sql)
if sql.startswith('select'):
rows = curs.fetchall()
curs.close()
return rows
else:
curs.execute('commit')
curs.close() def close(self):
"""
关闭数据库连接
"""
self.source_conn.close() def backup_table(self,table_schema,table_name,is_com):
"""
备份表
:param table_schema: 数据库schema
:param table_name: 表名称
:param is_com: 是否包含company_id
"""
if is_com>=1:
backup_cmd="mysqldump -u%s -p%s -h %s -P %s --single-transaction %s %s --where='company_id=1'>%s.sql"%(self.mysql_user,self.mysql_pwd,self.mysql_host,self.mysql_port,table_schema,table_name,table_name)
else:
backup_cmd = "mysqldump -u%s -p%s -h %s -P %s --single-transaction %s %s>%s.sql" % (
self.mysql_user, self.mysql_pwd, self.mysql_host,self.mysql_port, table_schema, table_name, table_name)
os.system(backup_cmd) def mysql_import(self,table_schema,table_name):
"""
数据导入
:param table_schema: 数据库schema
:param table_name: 表名称
"""
import_cmd='mysql -u%s -p%s -h %s -P %s -D %s -e"source %s.sql"'%(self.dest_user,self.dest_pwd,self.dest_host,self.dest_port,table_schema,table_name)
print(import_cmd)
os.system(import_cmd) def run(self):
"""
执行入口
:return:
"""
#从源端获取库 表名称
sql="select table_schema,table_name,sum(case when column_name='company_id' then 1 else 0 end) company_id_count from information_schema.`COLUMNS` where TABLE_SCHEMA not in ('mysql','information_schema','performance_schema','sys') group by table_schema,table_name"
rows=self.execute_sql(self.source_conn,sql)
for row in rows:
table_schema=row[0]
table_name=row[1]
is_comid=row[2]
#数据备份
self.backup_table(table_schema,table_name,is_comid)
#数据导入
self.mysql_import(table_schema,table_name) if __name__ == '__main__':
rsync_data=RsyncData()
rsync_data.run()
rsync_data.close()
最新文章
- AEAI CRM_v1.5.2升级说明，开源客户关系管理系统
- css水平垂直居中对齐方式
- mongoose 和 mongoDB
- dede使用方法---用js让当前导航高亮显示
- Java使用Jetty实现嵌入式Web服务器及Servlet容器
- ETM and PTM
- [NOI2001]反正切函数的应用
- NIC Bonding: 2 nic port as 1 interface
- OpenCV1.0在VC ++6.0下的配置
- spring 注解方式配置Bean
- [CI]CodeIgniter视图 & 模型 & 控制器
- C# 算法系列一基本数据结构
- 查看运行中的Java其配置的堆大小
- SparkStreaming 运行原理与核心概念
- django简介及URL
- win7下设置挂载Linux服务器nfs共享的数据 -- 转
- electron的通信
- JaveScript初学-基础语法
- tensorflow: a Implementation of rotation ops (旋转的函数实现方法)
- Mac OSX下Go语言开发环境的搭建与配置--使用InteliJ IDEA 13