不同数据库进行同步和增量数据（SQL server 与MySQL数据库为例） ...

场景

最近在做的一个项目需要将远程服务器的SQL server数据库中表的数据传输到本机的MySQL数据库中，而且远程的SQL server数据库表的数据会实时进行更新，而且差不多是一分钟内传输18条数据，例如现在是2023-12-4 15:09，在15:08这个时间内有18条数据需要首先辈SQL server数据库，再更新到MySQL数据库中，这种场景如果每分钟都能将18条数据放入SQL server数据库的话就非常简单了，但是在15：08的时候，这18条数据大概只来11条，剩下的7条大概在15:09或背面的时间陆续过来。我开始的想法是通过末了更新的时间的时间戳来查询新来的数据然后更新到MySQL中，但是由于在最终的时间内还会来前面时间的数据，这样会导致前面时间的数据丢失，所以我想了另外一方法。

• 首先利用python写一个步调来同步SQL sever的历史数据到MySQL数据库中
  
• 在SQL server中创建一个中间表。
  
• 在SQL server中要传输的表中创建一个触发器，当这个表更新数据则触发将更新的数据放入到中间表中
  
• 在python脚本中写一个循环来定期查抄中间表，我的SQL server表中由两个主键界说一条数据，所以中间表也是由两个字段界说一条数据，由于入库历史数据的数据量非常大，有几十万条，在这个入库历史数据的时间段内更新了许多条数据，所以大概中间表的数据与入库到MySQL中的字段有重复，所以我需要先验证中间表中的数据MySQL是否存在。
  
  
  • 存在则删除中间表中这条数据
    
  • 不存在则插入MySQL后删除这条数据
    
  
  
• 末了完成了入库步调，经过验证没有数据丢失
  

1.历史数据入库

历史数据入库我利用的python写的，首先界说两个数据库的信息

1. # 使用示例
2. sql_server_conn_params = {
3.     'driver': '{SQL Server}',
4.     'server': 'ip',
5.     'database': '数据库名',
6.     'uid': 'jzyg',
7.     'pwd': ''
8. }
10. mysql_conn_params = {
11.     'host': 'localhost',
12.     'user': 'root',
13.     'password': '123456',
14.     'database': '数据库名',
15.     'charset': 'utf8mb4'
16. }

复制代码

界说查询语句

1. querySolar = 'SELECT dtime,stationID,staionName,electric,tiltSolar,levelSolar,scatterSolar,directSolar,tiltSolar_day,levelSolar_day,scatterSolar_day,directSolar_day,sunShine_day FROM realData_Solar'

复制代码

界说入库历史数据函数

1.     def transfer_wind_data(self):
2.         # 连接到 SQL Server
3.         with pyodbc.connect(**self.sql_server_conn_params) as sql_server_conn:
4.             with sql_server_conn.cursor() as sql_server_cursor:
5.                 # 修改查询，仅选择上次同步后的数据
6.                 modified_query = self.queryWind + " WHERE dtime >= ? ORDER BY dtime"
7.                 sql_server_cursor.execute(modified_query, (self.wind_last_dtime,))
8.                 rows = sql_server_cursor.fetchall()
9.         if not rows:
10.             return  # 没有新数据
11.         # 连接到 MySQL
12.         with pymysql.connect(**self.mysql_conn_params) as mysql_conn:
13.             with mysql_conn.cursor() as mysql_cursor:
14.                 data_list = []
15.                 for row in rows:
16.                     observe_time = row[0].strftime("%Y-%m-%d %H:%M:00")
17.                     fsz_id = row[1].replace('"', "").strip()
18.                     station_name = row[2]
19.                     farmName = station_name.split("-")[0]
20.                     # 根据电站名查询电站号
21.                     sql = "SELECT FARMID FROM com_farm WHERE FARMNAME LIKE %s"
22.                     mysql_cursor.execute(sql, ('%' + farmName + '%',))
23.                     result = mysql_cursor.fetchone()
24.                     farm_id = None
25.                     if result is not None:
26.                         farm_id = result
27.                     # 处理查询结果为空的情况
28.                     if farm_id is not None:
29.                         farm_id = farm_id[0]
30.                     staion_name =  row[2]
31.                     wind_direction_instant = row[3]
32.                     wind_speed_instant = row[4]
33.                     wind_speed_two_min = row[5]
34.                     wind_speed_ten_min = row[6]
35.                     data = (observe_time,fsz_id,staion_name,farm_id,wind_direction_instant,
36.                             wind_speed_instant,wind_speed_two_min,wind_speed_ten_min)
37.                     data_list.append(data)
38.                     self.wind_last_dtime = row[0].strftime("%Y-%m-%d %H:%M:%S")
39.                 if data_list:
40.                     result = mysql_cursor.executemany('INSERT INTO wind_monitor'
41.                                                       '(observe_time,fsz_id,station_name,farm_id,wind_direction_instant,wind_speed_instant,wind_speed_two_min,wind_speed_ten_min) '
42.                                                       'VALUES (%s,%s,%s,%s,%s,%s,%s,%s)', data_list)  # 根据你的表结构修改
43.                     mysql_conn.commit()
44.                     print(f'wind_monitor表插入了{result}行数据.' if result else '没有新数据插入。')

复制代码

2.创建中间表和触发器

创建中间表

1. CREATE TABLE intermediateData_Wind AS SELECT * FROM realData_Wind WHERE 1=0;

复制代码

创建触发器

1. CREATE TRIGGER CopyToIntermediateTable
2. ON realData_Wind
3. AFTER INSERT
4. AS
5. BEGIN
6.     -- 插入操作
7.     INSERT INTO intermediateData_Wind (dtime, stationID, staionName, windDirectionInstant, windSpeedInstant, windSpeed2min, windSpeed10min)
8.     SELECT dtime, stationID, staionName, windDirectionInstant, windSpeedInstant, windSpeed2min, windSpeed10min
9.     FROM inserted;
10. END;

复制代码

3.创建轮询中间表代码

1. def transfer_insert_intermediateData_Wind(self):
2.     # 连接到 SQL Server
3.     with pyodbc.connect(**self.sql_server_conn_params) as sql_server_conn:
4.         with sql_server_conn.cursor() as sql_server_cursor:
5.             # 查询中间表中所有数据
6.             sql_server_cursor.execute(self.queryIntermediateData)
7.             intermediate_rows = sql_server_cursor.fetchall()
9.     # 用于跟踪删除和插入的数量
10.     deleted_count = 0
11.     inserted_count = 0
13.     # 连接到 MySQL
14.     with pymysql.connect(**self.mysql_conn_params) as mysql_conn:
15.         with mysql_conn.cursor() as mysql_cursor:
16.             for row in intermediate_rows:
17.                 observe_time = row[0].strftime("%Y-%m-%d %H:%M:00")
18.                 fsz_id = row[1].replace('"', "").strip()
20.                 # 检查wind_monitor表中是否存在相同数据
21.                 check_query = "SELECT COUNT(*) FROM wind_monitor WHERE observe_time = %s AND fsz_id = %s"
22.                 mysql_cursor.execute(check_query, (observe_time, fsz_id))
23.                 count = mysql_cursor.fetchone()[0]
24.                 dtime = row[0].strftime("%Y-%m-%d %H:%M:00.000")
25.                 if count > 0:
27.                     # 数据存在，从中间表删除
28.                     delete_query = "DELETE FROM intermediateData_Wind WHERE dtime = ? AND stationID = ?"
29.                     sql_server_cursor.execute(delete_query, (dtime, row[1]))
30.                     sql_server_conn.commit()
31.                     deleted_count += 1
32.                 else:
33.                     # 数据不存在，插入到wind_monitor并从中间表删除
34.                     station_name = row[2]
35.                     farmName = station_name.split("-")[0]
36.                     # 根据电站名查询电站号
37.                     farm_query = "SELECT FARMID FROM com_farm WHERE FARMNAME LIKE %s"
38.                     mysql_cursor.execute(farm_query, ('%' + farmName + '%',))
39.                     farm_result = mysql_cursor.fetchone()
40.                     farm_id = farm_result[0] if farm_result else None
42.                     wind_direction_instant = row[3]
43.                     wind_speed_instant = row[4]
44.                     wind_speed_two_min = row[5]
45.                     wind_speed_ten_min = row[6]
46.                     insert_data = (observe_time, fsz_id, station_name, farm_id, wind_direction_instant,
47.                                    wind_speed_instant, wind_speed_two_min, wind_speed_ten_min)
49.                     insert_query = 'INSERT INTO wind_monitor (observe_time, fsz_id, station_name, farm_id, wind_direction_instant, wind_speed_instant, wind_speed_two_min, wind_speed_ten_min) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)'
50.                     mysql_cursor.execute(insert_query, insert_data)
51.                     mysql_conn.commit()
52.                     inserted_count += 1
53.                     delete_query = "DELETE FROM intermediateData_Wind WHERE dtime = ? AND stationID = ?"
54.                     sql_server_cursor.execute(delete_query, (dtime, row[1]))
55.                     sql_server_conn.commit()
57.     # 打印删除和插入的数据统计
58.     print(f"从intermediateData_Wind表中删除了{deleted_count}条数据.")
59.     print(f"向wind_monitor表中插入了{inserted_count}条数据.")

复制代码

4.总体代码

1. import threadingimport timeimport pyodbcimport pymysqlclass DataTransfer:    def __init__(self, sql_server_conn_params, mysql_conn_params, queryWind,queryIntermediateData, interval=1):        self.sql_server_conn_params = sql_server_conn_params        self.mysql_conn_params = mysql_conn_params        self.queryWind = queryWind        self.queryIntermediateData = queryIntermediateData        self.interval = interval        self.wind_last_dtime = '1970-01-01 00:00:00'  # 初始时间    def clear_mysql_tables(self):        """清空 MySQL 中的指定表格数据"""        try:            with pymysql.connect(**self.mysql_conn_params) as mysql_conn:                with mysql_conn.cursor() as cursor:                    # 清空 wind_monitor 表                    cursor.execute("TRUNCATE TABLE wind_monitor")                    mysql_conn.commit()                    print("已清空 wind_monitor 表的数据。")        except Exception as e:            print(f"清空表格时发生错误: {e}")    def transfer_data(self):        self.transfer_wind_data()        while True:            try:                self.transfer_insert_intermediateData_Wind()            except Exception as e:                print(f"发生错误: {e}")            # 等待一定时间再次传输数据            time.sleep(self.interval)    def transfer_insert_intermediateData_Wind(self):        # 连接到 SQL Server        with pyodbc.connect(**self.sql_server_conn_params) as sql_server_conn:            with sql_server_conn.cursor() as sql_server_cursor:                # 查询中间表中所有数据                sql_server_cursor.execute(self.queryIntermediateData)                intermediate_rows = sql_server_cursor.fetchall()        # 用于跟踪删除和插入的数量        deleted_count = 0        inserted_count = 0        # 连接到 MySQL        with pymysql.connect(**self.mysql_conn_params) as mysql_conn:            with mysql_conn.cursor() as mysql_cursor:                for row in intermediate_rows:                    observe_time = row[0].strftime("%Y-%m-%d %H:%M:00")                    fsz_id = row[1].replace('"', "").strip()                    # 查抄wind_monitor表中是否存在相同数据                    check_query = "SELECT COUNT(*) FROM wind_monitor WHERE observe_time = %s AND fsz_id = %s"                    mysql_cursor.execute(check_query, (observe_time, fsz_id))                    count = mysql_cursor.fetchone()[0]                    dtime = row[0].strftime("%Y-%m-%d %H:%M:00.000")                    if count > 0:                        # 数据存在，从中间表删除                        delete_query = "DELETE FROM intermediateData_Wind WHERE dtime = ? AND stationID = ?"                        sql_server_cursor.execute(delete_query, (dtime, row[1]))                        sql_server_conn.commit()                        deleted_count += 1                    else:                        # 数据不存在，插入到wind_monitor并从中间表删除                        station_name = row[2]                        farmName = station_name.split("-")[0]                        # 根据电站名查询电站号                        farm_query = "SELECT FARMID FROM com_farm WHERE FARMNAME LIKE %s"                        mysql_cursor.execute(farm_query, ('%' + farmName + '%',))                        farm_result = mysql_cursor.fetchone()                        farm_id = farm_result[0] if farm_result else None                        wind_direction_instant = row[3]                        wind_speed_instant = row[4]                        wind_speed_two_min = row[5]                        wind_speed_ten_min = row[6]                        insert_data = (observe_time, fsz_id, station_name, farm_id, wind_direction_instant,                                       wind_speed_instant, wind_speed_two_min, wind_speed_ten_min)                        insert_query = 'INSERT INTO wind_monitor (observe_time, fsz_id, station_name, farm_id, wind_direction_instant, wind_speed_instant, wind_speed_two_min, wind_speed_ten_min) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)'                        mysql_cursor.execute(insert_query, insert_data)                        mysql_conn.commit()                        inserted_count += 1                        delete_query = "DELETE FROM intermediateData_Wind WHERE dtime = ? AND stationID = ?"                        sql_server_cursor.execute(delete_query, (dtime, row[1]))                        sql_server_conn.commit()        # 打印删除和插入的数据统计        print(f"从intermediateData_Wind表中删除了{deleted_count}条数据.")        print(f"向wind_monitor表中插入了{inserted_count}条数据.")    def transfer_wind_data(self):
2.         # 连接到 SQL Server
3.         with pyodbc.connect(**self.sql_server_conn_params) as sql_server_conn:
4.             with sql_server_conn.cursor() as sql_server_cursor:
5.                 # 修改查询，仅选择上次同步后的数据
6.                 modified_query = self.queryWind + " WHERE dtime >= ? ORDER BY dtime"
7.                 sql_server_cursor.execute(modified_query, (self.wind_last_dtime,))
8.                 rows = sql_server_cursor.fetchall()
9.         if not rows:
10.             return  # 没有新数据
11.         # 连接到 MySQL
12.         with pymysql.connect(**self.mysql_conn_params) as mysql_conn:
13.             with mysql_conn.cursor() as mysql_cursor:
14.                 data_list = []
15.                 for row in rows:
16.                     observe_time = row[0].strftime("%Y-%m-%d %H:%M:00")
17.                     fsz_id = row[1].replace('"', "").strip()
18.                     station_name = row[2]
19.                     farmName = station_name.split("-")[0]
20.                     # 根据电站名查询电站号
21.                     sql = "SELECT FARMID FROM com_farm WHERE FARMNAME LIKE %s"
22.                     mysql_cursor.execute(sql, ('%' + farmName + '%',))
23.                     result = mysql_cursor.fetchone()
24.                     farm_id = None
25.                     if result is not None:
26.                         farm_id = result
27.                     # 处理查询结果为空的情况
28.                     if farm_id is not None:
29.                         farm_id = farm_id[0]
30.                     staion_name =  row[2]
31.                     wind_direction_instant = row[3]
32.                     wind_speed_instant = row[4]
33.                     wind_speed_two_min = row[5]
34.                     wind_speed_ten_min = row[6]
35.                     data = (observe_time,fsz_id,staion_name,farm_id,wind_direction_instant,
36.                             wind_speed_instant,wind_speed_two_min,wind_speed_ten_min)
37.                     data_list.append(data)
38.                     self.wind_last_dtime = row[0].strftime("%Y-%m-%d %H:%M:%S")
39.                 if data_list:
40.                     result = mysql_cursor.executemany('INSERT INTO wind_monitor'
41.                                                       '(observe_time,fsz_id,station_name,farm_id,wind_direction_instant,wind_speed_instant,wind_speed_two_min,wind_speed_ten_min) '
42.                                                       'VALUES (%s,%s,%s,%s,%s,%s,%s,%s)', data_list)  # 根据你的表结构修改
43.                     mysql_conn.commit()
44.                     print(f'wind_monitor表插入了{result}行数据.' if result else '没有新数据插入。')
45.     def start(self):        # 在启动线程前先清空表格        self.clear_mysql_tables()        thread = threading.Thread(target=self.transfer_data)        thread.start()sql_server_conn_params = {    'driver': '{SQL Server}',    'server': '',    'database': '',    'uid': '',    'pwd': ''}mysql_conn_params = {    'host': 'localhost',    'user': 'root',    'password': '123456',    'database': '',    'charset': 'utf8mb4'}queryIntermediateData = "SELECT dtime,stationID,staionName,windDirectionInstant,windSpeedInstant,windSpeed2min,windSpeed10min FROM intermediateData_Wind"queryWind = 'SELECT dtime,stationID,staionName,windDirectionInstant,windSpeedInstant,windSpeed2min,windSpeed10min FROM realData_Wind'data_transfer = DataTransfer(sql_server_conn_params, mysql_conn_params, queryWind,queryIntermediateData)data_transfer.start()

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。