• 售前

  • 售后

热门帖子
入门百科

python实现MySQL指定表增量同步数据到clickhouse的脚本

[复制链接]
娣玲 显示全部楼层 发表于 2021-10-25 20:09:33 |阅读模式 打印 上一主题 下一主题
python实现MySQL指定表增量同步数据到clickhouse,脚本如下:
  1. #!/usr/bin/env python3
  2. # _*_ coding:utf8 _*_
  3. from pymysqlreplication import BinLogStreamReader
  4. from pymysqlreplication.row_event import (DeleteRowsEvent,UpdateRowsEvent,WriteRowsEvent,)
  5. import clickhouse_driver
  6. import configparser
  7. import os
  8. configfile='repl.ini'
  9. ########## 配置文件repl.ini 操作 ##################
  10. def create_configfile(configfile,log_file,log_pos):
  11.   config = configparser.ConfigParser()
  12.   if not os.path.exists(configfile):
  13.     config['replinfo'] = {'log_file':log_file,'log_pos':str(log_pos)}
  14.     with open(configfile,'w+') as f:
  15.       config.write(f)
  16. ### repl.ini 写操作 ##################
  17. def write_config(configfile,log_file,log_pos):
  18.   config = configparser.ConfigParser()
  19.   config.read(configfile)
  20.   config.set('replinfo','log_file',log_file)
  21.   config.set('replinfo','log_pos',str(log_pos))
  22.   if os.path.exists(configfile):
  23.     with open(configfile,'w+') as f:
  24.       config.write(f)
  25.   else:
  26.     create_configfile(configfile)
  27. ### 配置文件repl.ini 读操作 ##################
  28. def read_config(configfile):
  29.   config = configparser.ConfigParser()
  30.   config.read(configfile)
  31.   # print(config['replinfo']['log_file'])
  32.   # print(config['replinfo']['log_pos'])
  33.   return (config['replinfo']['log_file'],int(config['replinfo']['log_pos']))
  34. ############# clickhouse 操作 ##################
  35. def ops_clickhouse(db,table,sql):
  36.   column_type_dic={}
  37.   try:
  38.     client = clickhouse_driver.Client(host='127.0.0.1',\
  39.                      port=9000,\
  40.                      user='default',\
  41.                      password='clickhouse')
  42.     # sql="select name,type from system.columns where database='{0}' and table='{1}'".format(db,table)
  43.     client.execute(sql)
  44.   except Exception as error:
  45.     message = "获取clickhouse里面的字段类型错误. %s" % (error)
  46.     # logger.error(message)
  47.     print(message)
  48.     exit(1)
  49. MYSQL_SETTINGS = {'host':'127.0.0.1','port':13306,'user':'root','passwd':'Root@0101'}
  50. only_events=(DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent)
  51. def main():
  52.   ## 每次重启时,读取上次同步的log_file,log_pos
  53.   (log_file,log_pos) = read_config(configfile)
  54.   # print(log_file+'|'+ str(log_pos))
  55.   print('-----------------------------------------------------------------------------')
  56.   stream = BinLogStreamReader(connection_settings=MYSQL_SETTINGS, resume_stream=True, blocking=True, \
  57.                 server_id=10,
  58.                  only_tables='t_repl', only_schemas='test', \
  59.                 log_file=log_file,log_pos=log_pos, \
  60.                 only_events=only_events, \
  61.                 fail_on_table_metadata_unavailable=True, slave_heartbeat=10)
  62.   try:
  63.     for binlogevent in stream:
  64.       for row in binlogevent.rows:
  65.         ## delete操作
  66.         if isinstance(binlogevent, DeleteRowsEvent):
  67.           info = dict(row["values"].items())
  68.           # print("DELETE FROM `%s`.`%s` WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]) )
  69.           # print("ALTER TABLE `%s`.`%s` DELETE WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key]) )
  70.           sql="ALTER TABLE `%s`.`%s` DELETE WHERE %s = %s ;" %(binlogevent.schema ,binlogevent.table,binlogevent.primary_key,info[binlogevent.primary_key])
  71.         ## update 操作
  72.         elif isinstance(binlogevent, UpdateRowsEvent):
  73.           info_before = dict(row["before_values"].items())
  74.           info_after = dict(row["after_values"].items())
  75.           # info_set = str(info_after).replace(":","=").replace("{","").replace("}","")
  76.           info_set = str(info_after).replace(":", "=").replace("{", "").replace("}", "").replace("'","")
  77.           # print("UPDATE `%s`.`%s` SET %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key]  ) )
  78.           # print("ALTER TABLE %s.%s UPDATE %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key]  ) )
  79.           sql = "ALTER TABLE %s.%s UPDATE %s WHERE %s = %s ;"%(binlogevent.schema,binlogevent.table,info_set,binlogevent.primary_key,info_before[binlogevent.primary_key]  )
  80.         ## insert 操作
  81.         elif isinstance(binlogevent, WriteRowsEvent):
  82.           info = dict(row["values"].items())
  83.           # print("INSERT INTO %s.%s(%s)VALUES%s ;"%(binlogevent.schema,binlogevent.table , ','.join(info.keys()) ,str(tuple(info.values())) ) )
  84.           sql = "INSERT INTO %s.%s(%s)VALUES%s ;"%(binlogevent.schema,binlogevent.table , ','.join(info.keys()) ,str(tuple(info.values())) )
  85.         ops_clickhouse('test', 't_repl',sql )
  86.         # 当前log_file,log_pos写入配置文件
  87.         write_config(configfile, stream.log_file, stream.log_pos)
  88.   except Exception as e:
  89.     print(e)
  90.   finally:
  91.     stream.close()
  92. if __name__ == "__main__":
  93.   main()
  94. '''
  95. BinLogStreamReader()参数
  96. ctl_connection_settings:集群保存模式信息的连接设置
  97. resume_stream:从位置或binlog的最新事件或旧的可用事件开始
  98. log_file:设置复制开始日志文件
  99. log_pos:设置复制开始日志pos(resume_stream应该为true)
  100. auto_position:使用master_auto_position gtid设置位置
  101. blocking:在流上读取被阻止
  102. only_events:允许的事件数组
  103. ignored_events:被忽略的事件数组
  104. only_tables:包含要观看的表的数组(仅适用于binlog_format ROW)
  105. ignored_tables:包含要跳过的表的数组
  106. only_schemas:包含要观看的模式的数组
  107. ignored_schemas:包含要跳过的模式的数组
  108. freeze_schema:如果为true,则不支持ALTER TABLE。速度更快。
  109. skip_to_timestamp:在达到指定的时间戳之前忽略所有事件。
  110. report_slave:在SHOW SLAVE HOSTS中报告奴隶。
  111. slave_uuid:在SHOW SLAVE HOSTS中报告slave_uuid。
  112. fail_on_table_metadata_unavailable:如果我们无法获取有关row_events的表信息,应该引发异常
  113. slave_heartbeat:(秒)主站应主动发送心跳连接。这也减少了复制恢复时GTID复制的流量(在许多事件在binlog中跳过的情况下)。请参阅mysql文档中的MASTER_HEARTBEAT_PERIOD以了解语义
  114. '''
复制代码
知识点扩展:
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增量同步数据内容请搜索草根技术分享从前的文章或继续欣赏下面的相关文章盼望各人以后多多支持草根技术分享!

帖子地址: 

回复

使用道具 举报

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

本版积分规则

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

  • 微信公众号

  • 商务合作