python实现MySQL指定表增量同步数据到clickhouse的脚本python实现MySQL指定表增量同步数据到clickhouse,脚本如下:
#!/usr/bin/env python3
# _*_ coding:utf8 _*_
from pymysqlreplication import BinLogStreamReader
w_event import (DeleteRowsEvent,UpdateRowsEvent,WriteRowsEvent,)
import clickhouse_driver
import configparser
import os
configfile='repl.ini'
>> 配置⽂件repl.ini 操作 >>>###
def create_configfile(configfile,log_file,log_pos):
config = configparser.ConfigParser()
if not ists(configfile):
config['replinfo'] = {'log_file':log_file,'log_pos':str(log_pos)}
with open(configfile,'w+') as f:
config.write(f)
### repl.ini 写操作 >>>###
def write_config(configfile,log_file,log_pos):
config = configparser.ConfigParser()
config.set('replinfo','log_file',log_file)
config.set('replinfo','log_pos',str(log_pos))
if ists(configfile):
with open(configfile,'w+') as f:
config.write(f)
else:
create_configfile(configfile)
### 配置⽂件repl.ini 读操作 >>>###
def read_config(configfile):
config = configparser.ConfigParser()
# print(config['replinfo']['log_file'])
# print(config['replinfo']['log_pos'])
return (config['replinfo']['log_file'],int(config['replinfo']['log_pos']))
>>### clickhouse 操作 >>>###
def ops_clickhouse(db,table,sql):
column_type_dic={}
try:
client = clickhouse_driver.Client(host='127.0.0.1',\
port=9000,\
user='default',\
password='clickhouse')
drop table if exists admin# sql="select name,type lumns where database='{0}' and table='{1}'".format(db,table)
except Exception as error:
message = "获取clickhouse⾥⾯的字段类型错误. %s" % (error)
# (message)
print(message)
exit(1)
MYSQL_SETTINGS = {'host':'127.0.0.1','port':13306,'user':'root','passwd':'Root@0101'}
only_events=(DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent)
def main():
## 每次重启时,读取上次同步的log_file,log_pos
(log_file,log_pos) = read_config(configfile)
# print(log_file+'|'+ str(log_pos))
print('-----------------------------------------------------------------------------')
stream = BinLogStreamReader(connection_settings=MYSQL_SETTINGS, resume_stream=True, blocking=True, \
server_id=10,
only_tables='t_repl', only_schemas='test', \
log_file=log_file,log_pos=log_pos, \
only_events=only_events, \
fail_on_table_metadata_unavailable=True, slave_heartbeat=10)
try:
for binlogevent in stream:
for row ws:
## delete操作
if isinstance(binlogevent, DeleteRowsEvent):
info = dict(row["values"].items())
# print("DELETE FROM `%s`.`%s` WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]) )
# print("ALTER TABLE `%s`.`%s` DELETE WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]) )
sql="ALTER TABLE `%s`.`%s` DELETE WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key])
## update 操作
elif isinstance(binlogevent, UpdateRowsEvent):
info_before = dict(row["before_values"].items())
info_after = dict(row["after_values"].items())
# info_set = str(info_after).replace(":","=").replace("{","").replace("}","")
info_set = str(info_after).replace(":", "=").replace("{", "").replace("}", "").replace("'","")
# print("UPDATE `%s`.`%s` SET %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key] ) )
# print("ALTER TABLE %s.%s UPDATE %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key] ) ) sql = "ALTER TABLE %s.%s UPDATE %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key] ) ## insert 操作
elif isinstance(binlogevent, WriteRowsEvent):
info = dict(row["values"].items())
# print("INSERT INTO %s.%s(%s)VALUES%s ;"%(binlogevent.schema,binlogevent.table , ','.join(info.keys()) ,str(tuple(info.values())) ) )
sql = "INSERT INTO %s.%s(%s)VALUES%s ;"%(binlogevent.schema,binlogevent.table , ','.join(info.keys()) ,str(tuple(info.values())) )
ops_clickhouse('test', 't_repl',sql )
# 当前log_file,log_pos写⼊配置⽂件
write_config(configfile, stream.log_file, stream.log_pos)
except Exception as e:
print(e)
finally:
stream.close()
if __name__ == "__main__":
main()
'''
BinLogStreamReader()参数
ctl_connection_settings:集保存模式信息的连接设置
resume_stream:从位置或binlog的最新事件或旧的可⽤事件开始
log_file:设置复制开始⽇志⽂件
log_pos:设置复制开始⽇志pos(resume_stream应该为true)
auto_position:使⽤master_auto_position gtid设置位置
blocking:在流上读取被阻⽌
only_events:允许的事件数组
ignored_events:被忽略的事件数组
only_tables:包含要观看的表的数组(仅适⽤于binlog_format ROW)
ignored_tables:包含要跳过的表的数组
only_schemas:包含要观看的模式的数组
ignored_schemas:包含要跳过的模式的数组
freeze_schema:如果为true,则不⽀持ALTER TABLE。速度更快。
skip_to_timestamp:在达到指定的时间戳之前忽略所有事件。
report_slave:在SHOW SLAVE HOSTS中报告奴⾪。
slave_uuid:在SHOW SLAVE HOSTS中报告slave_uuid。
fail_on_table_metadata_unavailable:如果我们⽆法获取有关row_events的表信息,应该引发异常
slave_heartbeat:(秒)主站应主动发送⼼跳连接。这也减少了复制恢复时GTID复制的流量(在许多事件在binlog中跳过的情况下)。请参阅mysql⽂档中的MASTER_HEARTBEAT_PERIOD以了解语义'''
知识点扩展:
MySQL备份-增量同步
mysql增量同步主要使⽤binlog⽂件进⾏同步,binlog⽂件主要记录的是数据库更新操作相关的内容。
1. 备份数据的意义
针对不同业务,7*24⼩时提供服务和数据的重要性不同。
数据库数据是⽐较核⼼的数据,对企业的经营⾄关重要,数据库备份显得尤为重要。
2. 备份数据库
MySQL数据库⾃带的备份命令 `mysqldump`,基本使⽤⽅法:
语法:`mysqldump -u username -p password dbname > filename.sql`
执⾏备份命令
`mysqldump -uroot -pmysqladmin db_test > /opt/mysql_bak.sql`
查看备份内容
`grep -v "#|\*|--|^$" /opt/mysql_bak.sql`
到此这篇关于python实现MySQL指定表增量同步数据到clickhouse的脚本的⽂章就介绍到这了,更多相
关python实现MySQL增量同步数据内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持!
Windchill系统数据备份与恢复(建议)
« 上一篇
SQL用户自定义的数据类型、规则、默认
下一篇 »
热门文章
-
2021计算机二级考试预测题库及答案
2024-05-22 -
TPC产品介绍
2024-05-22 -
网络工程师学习笔记共享(共11章)
2024-05-22 -
计算机网络教学计划_工作计划
2024-05-22 -
2015-2016学年第二学期期末网络基础试卷及答案
2024-05-22 -
计算机网络基础(第一章)复习20计算机2班
2024-05-22 -
管理信息系统期末复习重点
2024-05-22 -
网络协议都有哪些
2024-05-22 -
常见的通信协议包括
2024-05-22 -
通信常用协议包括
2024-05-22 -
就业协议书包括的内容
2024-05-22 -
常用的安全协议包括
2024-05-22 -
协议书的基本内容包括( )。
2024-05-22 -
展会合作协议书
2024-05-22 -
《建设工程施工合同(示范文本)》(GF-2017-0201)通用条款
2024-05-22 -
快递承包合同协议书
2024-05-22 -
服务外包合同(通用8篇)
2024-05-22 -
通讯协议程序包括
2024-05-22 -
资料移交协议书通用模板【五篇】
2024-05-22 -
合同类型
2024-05-22
最新文章
-
计算机网络作业-网络嗅探器安装与使用
2024-05-22 -
本文把TCPIP讲绝了!
2024-05-22 -
同一个局域网内如何共享文件
2024-05-22 -
学士学位计算机试题及答案
2024-05-22 -
一个关于小程序与单片机的通信实例(TCPIP)
2024-05-22 -
计算机网络本科山大20年考试题库及答案
2024-05-22
发表评论