欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

python實(shí)現(xiàn)MySQL指定表增量同步數(shù)據(jù)到clickhouse的腳本

 更新時(shí)間:2021年02月26日 11:48:17   作者:xiaoxiao樹0  
這篇文章主要介紹了python實(shí)現(xiàn)MySQL指定表增量同步數(shù)據(jù)到clickhouse的腳本,本文通過實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下

python實(shí)現(xiàn)MySQL指定表增量同步數(shù)據(jù)到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里面的字段類型錯(cuò)誤. %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():
  ## 每次重啟時(shí),讀取上次同步的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 )
 
        # 當(dāng)前l(fā)og_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()參數(shù)
ctl_connection_settings:集群保存模式信息的連接設(shè)置
resume_stream:從位置或binlog的最新事件或舊的可用事件開始
log_file:設(shè)置復(fù)制開始日志文件
log_pos:設(shè)置復(fù)制開始日志pos(resume_stream應(yīng)該為true)
auto_position:使用master_auto_position gtid設(shè)置位置
blocking:在流上讀取被阻止
only_events:允許的事件數(shù)組
ignored_events:被忽略的事件數(shù)組
only_tables:包含要觀看的表的數(shù)組(僅適用于binlog_format ROW)
ignored_tables:包含要跳過的表的數(shù)組
only_schemas:包含要觀看的模式的數(shù)組
ignored_schemas:包含要跳過的模式的數(shù)組
freeze_schema:如果為true,則不支持ALTER TABLE。速度更快。
skip_to_timestamp:在達(dá)到指定的時(shí)間戳之前忽略所有事件。
report_slave:在SHOW SLAVE HOSTS中報(bào)告奴隸。
slave_uuid:在SHOW SLAVE HOSTS中報(bào)告slave_uuid。
fail_on_table_metadata_unavailable:如果我們無(wú)法獲取有關(guān)row_events的表信息,應(yīng)該引發(fā)異常
slave_heartbeat:(秒)主站應(yīng)主動(dòng)發(fā)送心跳連接。這也減少了復(fù)制恢復(fù)時(shí)GTID復(fù)制的流量(在許多事件在binlog中跳過的情況下)。請(qǐng)參閱mysql文檔中的MASTER_HEARTBEAT_PERIOD以了解語(yǔ)義
'''

知識(shí)點(diǎn)擴(kuò)展:

MySQL備份-增量同步

mysql增量同步主要使用binlog文件進(jìn)行同步,binlog文件主要記錄的是數(shù)據(jù)庫(kù)更新操作相關(guān)的內(nèi)容。

1. 備份數(shù)據(jù)的意義

針對(duì)不同業(yè)務(wù),7*24小時(shí)提供服務(wù)和數(shù)據(jù)的重要性不同。
數(shù)據(jù)庫(kù)數(shù)據(jù)是比較核心的數(shù)據(jù),對(duì)企業(yè)的經(jīng)營(yíng)至關(guān)重要,數(shù)據(jù)庫(kù)備份顯得尤為重要。

2. 備份數(shù)據(jù)庫(kù)

MySQL數(shù)據(jù)庫(kù)自帶的備份命令 `mysqldump`,基本使用方法:
語(yǔ)法:`mysqldump -u username -p password dbname > filename.sql`

執(zhí)行備份命令

`mysqldump -uroot -pmysqladmin db_test > /opt/mysql_bak.sql`

查看備份內(nèi)容

`grep -v "#|\*|--|^$" /opt/mysql_bak.sql`

到此這篇關(guān)于python實(shí)現(xiàn)MySQL指定表增量同步數(shù)據(jù)到clickhouse的腳本的文章就介紹到這了,更多相關(guān)python實(shí)現(xiàn)MySQL增量同步數(shù)據(jù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • 使用Python的Turtle繪制哆啦A夢(mèng)實(shí)例

    使用Python的Turtle繪制哆啦A夢(mèng)實(shí)例

    今天小編就為大家分享一篇使用Python的Turtle繪制哆啦A夢(mèng)實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來(lái)看看吧
    2019-11-11
  • Python中re模塊:匹配開頭/結(jié)尾(^/$)

    Python中re模塊:匹配開頭/結(jié)尾(^/$)

    本文主要介紹了Python中re模塊:匹配開頭/結(jié)尾(^/$),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2022-08-08
  • Python基于正則表達(dá)式實(shí)現(xiàn)文件內(nèi)容替換的方法

    Python基于正則表達(dá)式實(shí)現(xiàn)文件內(nèi)容替換的方法

    這篇文章主要介紹了Python基于正則表達(dá)式實(shí)現(xiàn)文件內(nèi)容替換的方法,涉及Python文件、目錄及字符串正則替換等相關(guān)操作技巧,需要的朋友可以參考下
    2017-08-08
  • Pyhton中防止SQL注入的方法

    Pyhton中防止SQL注入的方法

    這篇文章主要介紹了Pyhton中防止SQL注入的方法,本文講解的方法簡(jiǎn)單實(shí)用,需要的朋友可以參考下
    2015-02-02
  • Python?類中定義多個(gè)構(gòu)造器方法重載與泛方法

    Python?類中定義多個(gè)構(gòu)造器方法重載與泛方法

    這篇文章主要為大家介紹了Python?類中定義多個(gè)構(gòu)造器方法重載與泛方法示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪
    2023-03-03
  • Python模塊對(duì)Redis數(shù)據(jù)庫(kù)的連接與使用講解

    Python模塊對(duì)Redis數(shù)據(jù)庫(kù)的連接與使用講解

    這篇文章主要介紹了Python模塊對(duì)Redis數(shù)據(jù)庫(kù)的連接與使用,通過實(shí)例代碼給大家介紹了Python連接Redis數(shù)據(jù)庫(kù)方法,Python使用連接池連接Redis數(shù)據(jù)庫(kù)方法,感興趣的朋友跟隨小編一起看看吧
    2021-07-07
  • 深入理解Python虛擬機(jī)中整型(int)的實(shí)現(xiàn)原理及源碼剖析

    深入理解Python虛擬機(jī)中整型(int)的實(shí)現(xiàn)原理及源碼剖析

    在本篇文章當(dāng)中主要給大家介紹在 cpython 內(nèi)部是如何實(shí)現(xiàn)整型數(shù)據(jù) int 的,主要是分析 int 類型的表示方式,分析 int 類型的巧妙設(shè)計(jì)
    2023-03-03
  • Python項(xiàng)目管理Git常用命令詳圖講解

    Python項(xiàng)目管理Git常用命令詳圖講解

    這篇文章主要介紹了Python項(xiàng)目管理Git常用命令詳圖講解,文中附含詳細(xì)的圖片講解,建議收藏閱讀,有需要的朋友可以借鑒參考下,希望能夠有所幫助
    2021-09-09
  • Python中實(shí)現(xiàn)的RC4算法

    Python中實(shí)現(xiàn)的RC4算法

    這篇文章主要介紹了Python中實(shí)現(xiàn)的RC4算法,本文給出了類和函數(shù)兩種實(shí)現(xiàn)方式,需要的朋友可以參考下
    2015-02-02
  • Python文件系統(tǒng)模塊pathlib庫(kù)

    Python文件系統(tǒng)模塊pathlib庫(kù)

    這篇文章介紹了Python中的文件系統(tǒng)模塊pathlib庫(kù),文中通過示例代碼介紹的非常詳細(xì)。對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下
    2022-05-05

最新評(píng)論