python實現(xiàn)MySQL指定表增量同步數(shù)據(jù)到clickhouse的腳本
python實現(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里面的字段類型錯誤. %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 ) # 當前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:集群保存模式信息的連接設置 resume_stream:從位置或binlog的最新事件或舊的可用事件開始 log_file:設置復制開始日志文件 log_pos:設置復制開始日志pos(resume_stream應該為true) auto_position:使用master_auto_position gtid設置位置 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:在達到指定的時間戳之前忽略所有事件。 report_slave:在SHOW SLAVE HOSTS中報告奴隸。 slave_uuid:在SHOW SLAVE HOSTS中報告slave_uuid。 fail_on_table_metadata_unavailable:如果我們無法獲取有關row_events的表信息,應該引發(fā)異常 slave_heartbeat:(秒)主站應主動發(fā)送心跳連接。這也減少了復制恢復時GTID復制的流量(在許多事件在binlog中跳過的情況下)。請參閱mysql文檔中的MASTER_HEARTBEAT_PERIOD以了解語義 '''
知識點擴展:
MySQL備份-增量同步
mysql增量同步主要使用binlog文件進行同步,binlog文件主要記錄的是數(shù)據(jù)庫更新操作相關的內(nèi)容。
1. 備份數(shù)據(jù)的意義
針對不同業(yè)務,7*24小時提供服務和數(shù)據(jù)的重要性不同。
數(shù)據(jù)庫數(shù)據(jù)是比較核心的數(shù)據(jù),對企業(yè)的經(jīng)營至關重要,數(shù)據(jù)庫備份顯得尤為重要。
2. 備份數(shù)據(jù)庫
MySQL數(shù)據(jù)庫自帶的備份命令 `mysqldump`,基本使用方法:
語法:`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`
到此這篇關于python實現(xiàn)MySQL指定表增量同步數(shù)據(jù)到clickhouse的腳本的文章就介紹到這了,更多相關python實現(xiàn)MySQL增量同步數(shù)據(jù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
Python基于正則表達式實現(xiàn)文件內(nèi)容替換的方法
這篇文章主要介紹了Python基于正則表達式實現(xiàn)文件內(nèi)容替換的方法,涉及Python文件、目錄及字符串正則替換等相關操作技巧,需要的朋友可以參考下2017-08-08Python模塊對Redis數(shù)據(jù)庫的連接與使用講解
這篇文章主要介紹了Python模塊對Redis數(shù)據(jù)庫的連接與使用,通過實例代碼給大家介紹了Python連接Redis數(shù)據(jù)庫方法,Python使用連接池連接Redis數(shù)據(jù)庫方法,感興趣的朋友跟隨小編一起看看吧2021-07-07深入理解Python虛擬機中整型(int)的實現(xiàn)原理及源碼剖析
在本篇文章當中主要給大家介紹在 cpython 內(nèi)部是如何實現(xiàn)整型數(shù)據(jù) int 的,主要是分析 int 類型的表示方式,分析 int 類型的巧妙設計2023-03-03