### 使用 Python 实现两个数据库之间的定时数据同步 为了实现两个数据库间的定时数据同步，可以利用 `schedule` 库来设置任务调度，并借助 `pandas` 和特定的数据库驱动程序（如 `pyodbc` 或者 `mysql-connector-python`）来进行数据操作。下面展示了一个简单的例子，该实例假设源数据库为 MySQL 而目标数据库为 MS SQL Server。 #### 安装必要的库 首先安装所需的Python包： ```bash pip install schedule pandas pyodbc mysql-connector-python ``` #### 数据库连接配置 定义用于连接不同类型的数据库的函数，以便能够轻松切换不同的数据库环境。 ```python import pyodbc import mysql.connector as mysql def connect_mysql(): connection = mysql.connect( host='localhost', user='root', password='password', database='source_db' ) return connection def connect_mssql(): conn_str = ( r'DRIVER={ODBC Driver 17 for SQL Server};' r'SERVER=localhost;' r'DATABASE=target_db;' r'Trusted_Connection=yes;' # Use Windows Authentication ) connection = pyodbc.connect(conn_str) return connection ``` #### 同步逻辑编写 创建一个名为 `sync_data()` 的方法负责从源表提取最新记录并将其插入到目的表中。这里仅提供基本框架；实际应用时可能需要更复杂的处理机制以确保一致性和效率。 ```python import pandas as pd def sync_data(): try: src_conn = connect_mysql() dest_conn = connect_mssql() query = "SELECT * FROM source_table WHERE last_update >= (SELECT MAX(last_sync_time) FROM target_table)" df = pd.read_sql(query, con=src_conn) if not df.empty: insert_query = """ INSERT INTO target_table(column1, column2,...) VALUES (?, ?, ...) """ cursor = dest_conn.cursor() for index, row in df.iterrows(): values = tuple(row[col] for col in df.columns) cursor.execute(insert_query, values) dest_conn.commit() update_last_sync_time(dest_conn) # 更新最后同步时间 else: print("No new records found.") except Exception as e: print(f"Error occurred during data synchronization: {e}") finally: src_conn.close() dest_conn.close() def update_last_sync_time(connection): current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') sql = f""" UPDATE target_table SET last_sync_time='{current_time}' WHERE id IN (SELECT TOP(1) id FROM target_table ORDER BY id DESC); """ with connection.cursor() as cur: cur.execute(sql) connection.commit() ``` #### 设置定时任务 通过 `schedule` 来安排上述的数据同步过程定期运行。 ```python import time import schedule if __name__ == "__main__": # Schedule the job every day at midnight. schedule.every().day.at("00:00").do(sync_data) while True: schedule.run_pending() time.sleep(60) # Check once per minute whether it's time to run scheduled jobs ``` 此方案提供了基础架构，在真实场景下还需要考虑更多因素比如错误重试策略、并发控制以及性能优化等问题[^1]。