MySQL实现全量同步和增量同步到SQL Server或其他关系型库

在将MySQL中的两张表同步到SQL Server的过程中，全量同步和增量同步各有其优缺点。全量同步简单直接但可能耗时较长且资源消耗大，而增量同步则更加高效但需要额外的逻辑来处理数据的变更。以下是对这两种同步方式的详细解释及代码示例的完善。

完整代码示例

以下是一个完整的示例，包括全量同步和增量同步，以及使用schedule库来设置定时任务。

import pymysql
import pyodbc
from datetime import datetime, timedelta
import schedule
import time# MySQL 数据库连接函数
def get_mysql_connection():return pymysql.connect(host='localhost',         # 替换为你的 MySQL 服务器地址user='root',              # 替换为你的 MySQL 用户名password='password123',   # 替换为你的 MySQL 密码database='test_db'        # 替换为你的 MySQL 数据库名)# SQL Server 数据库连接函数
def get_sqlserver_connection():return pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};''SERVER=your_sqlserver_host;'  # 替换为你的 SQL Server 服务器地址或 IP'DATABASE=test_sqlserver_db;'  # 替换为你的 SQL Server 数据库名'UID=sqlserver_user;'          # 替换为你的 SQL Server 用户名'PWD=sqlserver_password123'    # 替换为你的 SQL Server 密码)# 全量同步函数
def full_sync_tables():mysql_conn = get_mysql_connection()sqlserver_conn = get_sqlserver_connection()try:mysql_cursor = mysql_conn.cursor()sqlserver_cursor = sqlserver_conn.cursor()# 清空SQL Server中的表数据sqlserver_cursor.execute("TRUNCATE TABLE table1")sqlserver_cursor.execute("TRUNCATE TABLE table2")# 从MySQL表中查询所有数据并插入到SQL Server表中for table in ['table1', 'table2']:mysql_cursor.execute(f"SELECT * FROM {table}")mysql_data = mysql_cursor.fetchall()columns = len(mysql_cursor.description)placeholders = ', '.join(['?'] * columns)insert_query = f"INSERT INTO {table} VALUES ({placeholders})"for row in mysql_data:sqlserver_cursor.execute(insert_query, row)sqlserver_conn.commit()finally:mysql_cursor.close()mysql_conn.close()sqlserver_cursor.close()sqlserver_conn.close()# 增量同步函数
def incremental_sync_tables(last_sync_time):mysql_conn = get_mysql_connection()sqlserver_conn = get_sqlserver_connection()try:mysql_cursor = mysql_conn.cursor()sqlserver_cursor = sqlserver_conn.cursor()# 获取MySQL中自上次同步以来的增量数据for table in ['table1', 'table2']:mysql_cursor.execute(f"SELECT * FROM {table} WHERE update_time > %s", (last_sync_time,))mysql_data = mysql_cursor.fetchall()# 插入或更新SQL Server中的数据update_query = f"UPDATE {table} SET {} WHERE id = ?".format(', '.join([f"{col} = ?" for col in mysql_cursor.description[1:]]))insert_query = f"INSERT INTO {table} ({}) VALUES ({})".format(', '.join([col[0] for col in mysql_cursor.description]),', '.join(['?'] * len(mysql_cursor.description)))for row in mysql_data:sqlserver_cursor.execute(f"SELECT id FROM {table} WHERE id = ?", (row[0],))result = sqlserver_cursor.fetchone()if result:sqlserver_cursor.execute(update_query, row[1:] + (row[0],))else:sqlserver_cursor.execute(insert_query, row)# 处理删除操作（假设MySQL有逻辑删除标记字段is_deleted）for table in ['table1', 'table2']:sqlserver_cursor.execute(f"SELECT id FROM {table}")sqlserver_ids = [row[0] for row in sqlserver_cursor.fetchall()]mysql_cursor.execute(f"SELECT id FROM {table} WHERE is_deleted = 1 AND update_time > %s", (last_sync_time,))deleted_ids = [row[0] for row in mysql_cursor.fetchall()]for id_ in set(sqlserver_ids) - set(deleted_ids):sqlserver_cursor.execute(f"DELETE FROM {table} WHERE id = ?", (id_,))sqlserver_conn.commit()finally:mysql_cursor.close()mysql_conn.close()sqlserver_cursor.close()sqlserver_conn.close()# 定时任务函数
def schedule_sync_tasks():# 每天凌晨1点进行全量同步schedule.every().day.at("01:00").do(full_sync_tables)# 每5分钟进行增量同步last_sync_time = datetime.now() - timedelta(minutes=5)  # 初始化为5分钟前，之后每次调用都会更新def run_incremental_sync():nonlocal last_sync_timeincremental_sync_tables(last_sync_time)last_sync_time = datetime.now()  # 更新上次同步时间schedule.every(5).minutes.do(run_incremental_sync)# 运行调度器while True:schedule.run_pending()time.sleep(1)# 运行定时任务
if __name__ == "__main__":schedule_sync_tasks()