本文目录导读：
我来详细说明如何编写数据同步校验脚本,涵盖不同场景下的实现方案。
数据同步校验的核心目标
主要验证点：
- 数据完整性：记录数是否一致
- 数据准确性：字段值是否相同
- 数据一致性：主键和外键关系是否完整
- 时效性：同步延迟是否在可接受范围
简单计数校验脚本（Python）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pymysql
import hashlib
import time
from datetime import datetime
class DataSyncValidator:
def __init__(self, source_config, target_config):
self.source_conn = pymysql.connect(**source_config)
self.target_conn = pymysql.connect(**target_config)
def count_check(self, table_name):
"""记录数校验"""
source_count = self._get_count(self.source_conn, table_name)
target_count = self._get_count(self.target_conn, table_name)
status = "PASS" if source_count == target_count else "FAIL"
print(f"[{status}] 表 {table_name}: 源库={source_count}, 目标库={target_count}")
return source_count == target_count
def _get_count(self, conn, table_name):
cursor = conn.cursor()
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
return cursor.fetchone()[0]
def checksum_check(self, table_name, primary_key, columns=None):
"""使用CHECKSUM进行数据一致性校验"""
if columns is None:
columns = "*"
# 源库校验和
source_checksum = self._get_checksum(self.source_conn, table_name, primary_key, columns)
target_checksum = self._get_checksum(self.target_conn, table_name, primary_key, columns)
status = "PASS" if source_checksum == target_checksum else "FAIL"
print(f"[{status}] 表 {table_name} Checksum: 源库={source_checksum}, 目标库={target_checksum}")
return source_checksum == target_checksum
def _get_checksum(self, conn, table_name, primary_key, columns):
cursor = conn.cursor()
if columns == "*":
cursor.execute(f"SELECT CHECKSUM(*) FROM {table_name}")
else:
cursor.execute(f"SELECT BIT_XOR(CRC32(CONCAT({columns}))) FROM {table_name}")
return cursor.fetchone()[0]
def close(self):
self.source_conn.close()
self.target_conn.close()
# 使用示例
if __name__ == "__main__":
# 数据库配置
source_config = {
'host': 'source_host',
'user': 'user',
'password': 'password',
'database': 'source_db',
'charset': 'utf8mb4'
}
target_config = {
'host': 'target_host',
'user': 'user',
'password': 'password',
'database': 'target_db',
'charset': 'utf8mb4'
}
validator = DataSyncValidator(source_config, target_config)
# 执行校验
tables = ['users', 'orders', 'products']
for table in tables:
validator.count_check(table)
validator.checksum_check(table, 'id', 'name,email,status')
validator.close()
完整数据对比脚本（支持分批处理）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pandas as pd
from sqlalchemy import create_engine
import logging
from datetime import datetime
class DataSyncComparer:
def __init__(self, source_conn_str, target_conn_str):
self.source_engine = create_engine(source_conn_str)
self.target_engine = create_engine(target_conn_str)
self.setup_logging()
def setup_logging(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('sync_validation.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def compare_tables(self, table_name, key_columns, batch_size=1000):
"""分批对比数据"""
source_count = self._get_count(self.source_engine, table_name)
target_count = self._get_count(self.target_engine, table_name)
self.logger.info(f"表 {table_name}: 源库={source_count}, 目标库={target_count}")
if source_count != target_count:
self.logger.error(f"记录数不匹配: {source_count} vs {target_count}")
return False
# 分批获取并比较数据
offset = 0
all_consistent = True
while offset < source_count:
source_data = self._get_batch_data(self.source_engine, table_name, key_columns, offset, batch_size)
target_data = self._get_batch_data(self.target_engine, table_name, key_columns, offset, batch_size)
# 按主键排序后比较
source_data = source_data.sort_values(by=key_columns).reset_index(drop=True)
target_data = target_data.sort_values(by=key_columns).reset_index(drop=True)
if not source_data.equals(target_data):
self.logger.error(f"数据不匹配: offset={offset}")
self._log_differences(source_data, target_data, key_columns)
all_consistent = False
offset += batch_size
self.logger.info(f"已校验 {min(offset, source_count)}/{source_count} 条记录")
return all_consistent
def _get_count(self, engine, table_name):
query = f"SELECT COUNT(*) FROM {table_name}"
return pd.read_sql(query, engine).iloc[0, 0]
def _get_batch_data(self, engine, table_name, key_columns, offset, limit):
query = f"""
SELECT * FROM {table_name}
ORDER BY {', '.join(key_columns)}
LIMIT {limit} OFFSET {offset}
"""
return pd.read_sql(query, engine)
def _log_differences(self, source_df, target_df, key_columns):
"""记录具体差异"""
# 找到不同行
differences = source_df != target_df
diff_rows = differences.any(axis=1)
for idx in diff_rows[diff_rows].index:
self.logger.error(f"差异记录 (行 {idx}):")
for col in source_df.columns:
if source_df.loc[idx, col] != target_df.loc[idx, col]:
self.logger.error(f" 字段 {col}: {source_df.loc[idx, col]} -> {target_df.loc[idx, col]}")
# 使用示例
if __name__ == "__main__":
source_conn = "mysql+pymysql://user:password@source_host/source_db"
target_conn = "mysql+pymysql://user:password@target_host/target_db"
comparer = DataSyncComparer(source_conn, target_conn)
# 比较多个表
tables_to_compare = [
{"name": "users", "keys": ["user_id"]},
{"name": "orders", "keys": ["order_id"]}
]
for table in tables_to_compare:
result = comparer.compare_tables(table["name"], table["keys"])
print(f"表 {table['name']}: {'通过' if result else '失败'}")
高级校验：支持多种数据源
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import redis
from elasticsearch import Elasticsearch
import json
class MultiSourceValidator:
def __init__(self):
self.validators = {
'mysql': self._validate_mysql,
'redis': self._validate_redis,
'elasticsearch': self._validate_elasticsearch
}
def validate(self, source_type, target_type, source_config, target_config, query_config):
"""多数据源校验"""
source_data = self._get_data(source_type, source_config, query_config)
target_data = self._get_data(target_type, target_config, query_config)
return self._compare_data(source_data, target_data, query_config.get('key_field'))
def _get_data(self, data_type, config, query_config):
if data_type == 'mysql':
return self._get_mysql_data(config, query_config)
elif data_type == 'redis':
return self._get_redis_data(config, query_config)
elif data_type == 'elasticsearch':
return self._get_es_data(config, query_config)
def _get_mysql_data(self, config, query_config):
conn = pymysql.connect(**config)
cursor = conn.cursor()
cursor.execute(query_config['query'])
return cursor.fetchall()
def _get_redis_data(self, config, query_config):
r = redis.Redis(**config)
keys = r.keys(query_config['pattern'])
data = {}
for key in keys:
data[key] = r.get(key)
return data
def _get_es_data(self, config, query_config):
es = Elasticsearch([config])
result = es.search(index=query_config['index'], body=query_config['query'])
return result['hits']['hits']
def _compare_data(self, source_data, target_data, key_field):
"""通用数据比较"""
inconsistencies = []
# 转换为字典便于比较
source_dict = {item[key_field]: item for item in source_data}
target_dict = {item[key_field]: item for item in target_data}
# 检查缺失和差异
for key in source_dict:
if key not in target_dict:
inconsistencies.append(f"目标缺少记录: {key}")
elif source_dict[key] != target_dict[key]:
inconsistencies.append(f"记录 {key} 数据不一致")
for key in target_dict:
if key not in source_dict:
inconsistencies.append(f"源缺少记录: {key}")
return inconsistencies
# 使用示例
validator = MultiSourceValidator()
# 比较MySQL和Redis中的数据
mysql_config = {
'host': 'localhost',
'user': 'root',
'password': 'password',
'database': 'test'
}
redis_config = {
'host': 'localhost',
'port': 6379
}
query_config = {
'query': 'SELECT user_id, name, email FROM users WHERE status = 1',
'pattern': 'user:*',
'key_field': 'user_id'
}
inconsistencies = validator.validate('mysql', 'redis', mysql_config, redis_config, query_config)
print(f"发现 {len(inconsistencies)} 处不一致")
Shell脚本实现（简单快速）
#!/bin/bash
# data_sync_check.sh
# 配置
SOURCE_HOST="source_host"
SOURCE_USER="root"
SOURCE_PASS="password"
SOURCE_DB="source_db"
TARGET_HOST="target_host"
TARGET_USER="root"
TARGET_PASS="password"
TARGET_DB="target_db"
# 表列表
TABLES=("users" "orders" "products")
check_table() {
local table=$1
echo "检查表: $table"
# 获取记录数
source_count=$(mysql -h $SOURCE_HOST -u $SOURCE_USER -p$SOURCE_PASS -D $SOURCE_DB -e "SELECT COUNT(*) FROM $table" | tail -1)
target_count=$(mysql -h $TARGET_HOST -u $TARGET_USER -p$TARGET_PASS -D $TARGET_DB -e "SELECT COUNT(*) FROM $table" | tail -1)
echo " 源库: $source_count"
echo " 目标库: $target_count"
if [ "$source_count" -eq "$target_count" ]; then
echo " ✓ 记录数一致"
# 检查校验和
source_checksum=$(mysql -h $SOURCE_HOST -u $SOURCE_USER -p$SOURCE_PASS -D $SOURCE_DB -e "SELECT CHECKSUM(*) FROM $table" | tail -1)
target_checksum=$(mysql -h $TARGET_HOST -u $TARGET_USER -p$TARGET_PASS -D $TARGET_DB -e "SELECT CHECKSUM(*) FROM $table" | tail -1)
if [ "$source_checksum" == "$target_checksum" ]; then
echo " ✓ 数据一致"
else
echo " ✗ 数据不一致"
return 1
fi
else
echo " ✗ 记录数不一致"
return 1
fi
}
# 主逻辑
echo "开始数据同步校验..."
failures=0
for table in "${TABLES[@]}"; do
if ! check_table $table; then
echo "表 $table 校验失败"
((failures++))
fi
echo ""
done
echo "校验完成: 成功 $(( ${#TABLES[@]} - $failures )), 失败 $failures"
exit $failures
定时校验（Crontab配置）
# 每天凌晨2点执行全量校验 0 2 * * * /path/to/data_sync_check.sh >> /var/log/sync_check.log 2>&1 # 每30分钟执行快速校验 */30 * * * * /path/to/quick_check.sh >> /var/log/quick_check.log 2>&1
关键建议
-
性能优化：
- 对大表使用分批处理
- 考虑使用并行处理
- 避免全表锁
-
错误处理：
- 实现重试机制
- 记录详细日志
- 发送告警通知
-
持续改进：
- 建立基线数据
- 自动化修复
- 监控趋势变化
根据你的具体需求选择合适的方案，MySQL和PostgreSQL场景推荐用Python脚本,简单的数量校验用Shell就够了。
