文章详情

短信预约信息系统项目管理师 报名、考试、查分时间动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2022-05-14 21:59

关注

python实现MySQL指定表增量同步数据到clickhouse,脚本如下:


#!/usr/bin/env python3
# _*_ coding:utf8 _*_
 
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_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 os.path.exists(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.read(configfile)
 
  config.set('replinfo','log_file',log_file)
  config.set('replinfo','log_pos',str(log_pos))
 
  if os.path.exists(configfile):
    with open(configfile,'w+') as f:
      config.write(f)
  else:
    create_configfile(configfile)
 
### 配置文件repl.ini 读操作 ##################
def read_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 操作 ##################
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')
    # sql="select name,type from system.columns where database='{0}' and table='{1}'".format(db,table)
    client.execute(sql)
 
  except Exception as error:
    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)
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 in binlogevent.rows:
        ## 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文件主要记录的是数据库更新操作相关的内容。

备份数据的意义

针对不同业务,7*24小时提供服务和数据的重要性不同。
数据库数据是比较核心的数据,对企业的经营至关重要,数据库备份显得尤为重要。

备份数据库

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增量同步数据内容请搜索自学编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持自学编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-数据库
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯