python实现MySQL指定表增量同步数据到clickhouse的脚本
python实现MySQL指定表增量同步数据到clickhouse,脚本如下:
#!/usr/bin/envpython3
#_*_coding:utf8_*_
frompymysqlreplicationimportBinLogStreamReader
frompymysqlreplication.row_eventimport(DeleteRowsEvent,UpdateRowsEvent,WriteRowsEvent,)
importclickhouse_driver
importconfigparser
importos
configfile='repl.ini'
##########配置文件repl.ini操作##################
defcreate_configfile(configfile,log_file,log_pos):
config=configparser.ConfigParser()
ifnotos.path.exists(configfile):
config['replinfo']={'log_file':log_file,'log_pos':str(log_pos)}
withopen(configfile,'w+')asf:
config.write(f)
###repl.ini写操作##################
defwrite_config(configfile,log_file,log_pos):
config=configparser.ConfigParser()
config.read(configfile)
config.set('replinfo','log_file',log_file)
config.set('replinfo','log_pos',str(log_pos))
ifos.path.exists(configfile):
withopen(configfile,'w+')asf:
config.write(f)
else:
create_configfile(configfile)
###配置文件repl.ini读操作##################
defread_config(configfile):
config=configparser.ConfigParser()
config.read(configfile)
#print(config['replinfo']['log_file'])
#print(config['replinfo']['log_pos'])
return(config['replinfo']['log_file'],int(config['replinfo']['log_pos']))
#############clickhouse操作##################
defops_clickhouse(db,table,sql):
column_type_dic={}
try:
client=clickhouse_driver.Client(host='127.0.0.1',\
port=9000,\
user='default',\
password='clickhouse')
#sql="selectname,typefromsystem.columnswheredatabase='{0}'andtable='{1}'".format(db,table)
client.execute(sql)
exceptExceptionaserror:
message="获取clickhouse里面的字段类型错误.%s"%(error)
#logger.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)
defmain():
##每次重启时，读取上次同步的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:
forbinlogeventinstream:
forrowinbinlogevent.rows:
##delete操作
ifisinstance(binlogevent,DeleteRowsEvent):
info=dict(row["values"].items())
#print("DELETEFROM`%s`.`%s`WHERE%s=%s;"%(binlogevent.schema,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]))
#print("ALTERTABLE`%s`.`%s`DELETEWHERE%s=%s;"%(binlogevent.schema,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]))
sql="ALTERTABLE`%s`.`%s`DELETEWHERE%s=%s;"%(binlogevent.schema,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key])
##update操作
elifisinstance(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%sWHERE%s=%s;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key]))
#print("ALTERTABLE%s.%sUPDATE%sWHERE%s=%s;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key]))
sql="ALTERTABLE%s.%sUPDATE%sWHERE%s=%s;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key])
##insert操作
elifisinstance(binlogevent,WriteRowsEvent):
info=dict(row["values"].items())
#print("INSERTINTO%s.%s(%s)VALUES%s;"%(binlogevent.schema,binlogevent.table,','.join(info.keys()),str(tuple(info.values()))))
sql="INSERTINTO%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)
exceptExceptionase:
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_positiongtid设置位置
blocking：在流上读取被阻止
only_events：允许的事件数组
ignored_events：被忽略的事件数组
only_tables：包含要观看的表的数组（仅适用于binlog_formatROW）
ignored_tables：包含要跳过的表的数组
only_schemas：包含要观看的模式的数组
ignored_schemas：包含要跳过的模式的数组
freeze_schema：如果为true，则不支持ALTERTABLE。速度更快。
skip_to_timestamp：在达到指定的时间戳之前忽略所有事件。
report_slave：在SHOWSLAVEHOSTS中报告奴隶。
slave_uuid：在SHOWSLAVEHOSTS中报告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-uusername-ppassworddbname>filename.sql`
执行备份命令
`mysqldump-uroot-pmysqladmindb_test>/opt/mysql_bak.sql`
查看备份内容
`grep-v"#|\*|--|^$"/opt/mysql_bak.sql`
到此这篇关于python实现MySQL指定表增量同步数据到clickhouse的脚本的文章就介绍到这了,更多相关python实现MySQL增量同步数据内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。