比较数据迁移后MySQL数据库和ClickHouse数据仓库中的表

设计一个MySQL数据库和Clickhouse数据仓库的表数据比较的详细程序流程，两张表是相同的结构，都有整型主键id字段，需要每次从数据库分批取得2000条数据，用于比较，比较操作的同时可以再取2000条数据，等上一次比较完成之后，开始比较，直到比较完所有的数据。比较操作需要比较两条两个数据库中，一个数据库有，另一个数据库没有的记录，以id是否存在作为标准，还需要比较两边都有记录的所有数据字段是否都相同，如果有不相同指出这条数据的id，以及不相同数据所在字段的名称和在两个数据库中的值，分批比较时需要考虑所有可能的情况。

以下是一个详细的设计方案，用于比较MySQL数据库和ClickHouse数据仓库中结构相同的表，支持并行数据获取和分批比较：

系统设计

1. 并行处理架构：

   • 主线程：负责数据比较
   • 工作线程：负责预取下一批数据
   • 使用队列实现线程间通信

2. 关键组件：

   • 数据获取器（DataFetcher）
   • 数据比较器（DataComparator）
   • 结果处理器（ResultHandler）
   • 批处理控制器（BatchController）

详细程序流程

import queue
import threading
from typing import Dict, List, Tuple, Optional# 假设已实现数据库连接池
from db_connectors import MySQLPool, ClickHousePoolclass BatchData:def __init__(self):self.mysql_data: Dict[int, dict] = {}  # {id: {field1:val1,...}}self.ch_data: Dict[int, dict] = {}self.last_id: int = 0  # 本批次最大IDclass DataComparator:def __init__(self, result_handler):self.result_handler = result_handlerdef compare_batch(self, batch: BatchData):# 1. 检查ID存在性差异mysql_ids = set(batch.mysql_data.keys())ch_ids = set(batch.ch_data.keys())# 1.1 MySQL有但ClickHouse缺失for mid in mysql_ids - ch_ids:self.result_handler.record_missing(mid, source="clickhouse")# 1.2 ClickHouse有但MySQL缺失for cid in ch_ids - mysql_ids:self.result_handler.record_missing(cid, source="mysql")# 2. 比较共有ID的字段值common_ids = mysql_ids & ch_idsfor cid in common_ids:mysql_row = batch.mysql_data[cid]ch_row = batch.ch_data[cid]self._compare_rows(cid, mysql_row, ch_row)def _compare_rows(self, id: int, mysql_row: dict, ch_row: dict):# 跳过主键字段fields = set(mysql_row.keys()) - {"id"}for field in fields:mysql_val = mysql_row[field]ch_val = ch_row[field]# 处理类型差异（如MySQL的Decimal转Float）if isinstance(mysql_val, Decimal):mysql_val = float(mysql_val)if isinstance(ch_val, Decimal):ch_val = float(ch_val)# 特殊处理浮点数精度if isinstance(mysql_val, float) and isinstance(ch_val, float):if abs(mysql_val - ch_val) > 1e-9:self.result_handler.record_mismatch(id, field, mysql_val, ch_val)elif mysql_val != ch_val:self.result_handler.record_mismatch(id, field, mysql_val, ch_val)class DataFetcher:BATCH_SIZE = 2000def __init__(self, mysql_pool, ch_pool):self.mysql_pool = mysql_poolself.ch_pool = ch_pooldef fetch_batch(self, last_id: int) -> Optional[BatchData]:batch = BatchData()# 从MySQL获取数据with self.mysql_pool.connection() as conn:cursor = conn.cursor(dictionary=True)cursor.execute("SELECT * FROM target_table ""WHERE id > %s ORDER BY id LIMIT %s",(last_id, self.BATCH_SIZE))for row in cursor:batch.mysql_data[row['id']] = rowbatch