如何编写数据同步校验脚本

本文目录导读：

1. 数据同步校验的核心目标
2. 简单计数校验脚本（Python）
3. 完整数据对比脚本（支持分批处理）
4. 高级校验：支持多种数据源
5. Shell脚本实现（简单快速）
6. 定时校验（Crontab配置）
7. 关键建议

我来详细说明如何编写数据同步校验脚本,涵盖不同场景下的实现方案。

数据同步校验的核心目标

主要验证点：

• 数据完整性：记录数是否一致
• 数据准确性：字段值是否相同
• 数据一致性：主键和外键关系是否完整
• 时效性：同步延迟是否在可接受范围

简单计数校验脚本（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

关键建议

1. 性能优化：

   • 对大表使用分批处理
   • 考虑使用并行处理
   • 避免全表锁

2. 错误处理：

   • 实现重试机制
   • 记录详细日志
   • 发送告警通知

3. 持续改进：

   • 建立基线数据
   • 自动化修复
   • 监控趋势变化

根据你的具体需求选择合适的方案，MySQL和PostgreSQL场景推荐用Python脚本,简单的数量校验用Shell就够了。