一、数据一致性概述二、迁移过程中的一致性风险2.1 一致性风险类型2.2 风险影响评估风险类型影响程度发生概率优先级数据丢失高中高数据重复中高高数据错误中中中数据不一致中中中完整性破坏高低高三、一致性保障策略3.1 事务保障事务迁移流程事务隔离级别隔离级别说明适用场景READ UNCOMMITTED读取未提交数据不重要数据READ COMMITTED读取已提交数据一般业务REPEATABLE READ可重复读重要业务SERIALIZABLE串行化核心业务3.2 幂等性保障幂等性设计幂等性代码示例import uuid class IdempotentMigration: def __init__(self, db): self.db db def generate_request_id(self): return str(uuid.uuid4()) def check_request_id(self, request_id): result self.db.execute( SELECT COUNT(*) FROM migration_requests WHERE request_id %s, (request_id,) ) return result.fetchone()[0] 0 def save_request_id(self, request_id): self.db.execute( INSERT INTO migration_requests (request_id, status, created_at) VALUES (%s, %s, NOW()), (request_id, processing) ) def update_request_status(self, request_id, status): self.db.execute( UPDATE migration_requests SET status %s, updated_at NOW() WHERE request_id %s, (status, request_id) ) def migrate_with_idempotency(self, data, request_idNone): if request_id is None: request_id self.generate_request_id() if self.check_request_id(request_id): print(fRequest {request_id} already processed) return {success: True, message: Already processed} try: self.save_request_id(request_id) with self.db.begin(): self._migrate_data(data) self.update_request_status(request_id, success) return {success: True, message: Migration successful} except Exception as e: self.update_request_status(request_id, failed) return {success: False, message: str(e)} def _migrate_data(self, data): for record in data: existing self.db.execute( SELECT id FROM target_table WHERE id %s, (record[id],) ).fetchone() if existing: self.db.execute( UPDATE target_table SET name %s, email %s WHERE id %s, (record[name], record[email], record[id]) ) else: self.db.execute( INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s), (record[id], record[name], record[email]) )3.3 分布式锁分布式锁实现Redis分布式锁代码示例import redis import time class DistributedLock: def __init__(self, redis_client, lock_key, expire_time30): self.redis redis_client self.lock_key lock_key self.expire_time expire_time self.lock_value None def acquire(self, timeout10): end_time time.time() timeout while time.time() end_time: self.lock_value str(uuid.uuid4()) result self.redis.set( self.lock_key, self.lock_value, nxTrue, exself.expire_time ) if result: return True time.sleep(0.1) return False def release(self): if self.lock_value: script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end self.redis.eval(script, 1, self.lock_key, self.lock_value) def __enter__(self): self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): self.release()3.4 数据校验点校验点机制校验点实现class CheckpointMigration: def __init__(self, db): self.db db def get_checkpoint(self, migration_id): result self.db.execute( SELECT position, status FROM migration_checkpoints WHERE migration_id %s, (migration_id,) ).fetchone() if result: return {position: result[0], status: result[1]} return {position: 0, status: pending} def set_checkpoint(self, migration_id, position, status): existing self.db.execute( SELECT id FROM migration_checkpoints WHERE migration_id %s, (migration_id,) ).fetchone() if existing: self.db.execute( UPDATE migration_checkpoints SET position %s, status %s, updated_at NOW() WHERE migration_id %s, (position, status, migration_id) ) else: self.db.execute( INSERT INTO migration_checkpoints (migration_id, position, status, created_at) VALUES (%s, %s, %s, NOW()), (migration_id, position, status) ) def run_migration(self, migration_id, data, batch_size1000): checkpoint self.get_checkpoint(migration_id) start_pos checkpoint[position] total_records len(data) print(fResuming migration from position {start_pos}) for i in range(start_pos, total_records, batch_size): batch data[i:ibatch_size] try: with self.db.begin(): self._process_batch(batch) self.set_checkpoint(migration_id, i len(batch), processing) print(fMigrated {i len(batch)}/{total_records} records) except Exception as e: print(fError at position {i}: {e}) self.set_checkpoint(migration_id, i, failed) raise self.set_checkpoint(migration_id, total_records, completed) print(Migration completed successfully) def _process_batch(self, batch): for record in batch: self.db.execute( INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE name %s, email %s, (record[id], record[name], record[email], record[name], record[email]) )四、数据校验机制4.1 校验类型4.2 数量校验数量校验流程数量校验代码示例class CountValidator: def __init__(self, source_db, target_db): self.source_db source_db self.target_db target_db def get_source_count(self, table_name): result self.source_db.execute(fSELECT COUNT(*) FROM {table_name}) return result.fetchone()[0] def get_target_count(self, table_name): result self.target_db.execute(fSELECT COUNT(*) FROM {table_name}) return result.fetchone()[0] def validate(self, table_name): source_count self.get_source_count(table_name) target_count self.get_target_count(table_name) is_valid source_count target_count return { table: table_name, source_count: source_count, target_count: target_count, is_valid: is_valid, difference: source_count - target_count }4.3 内容校验内容校验流程内容校验代码示例import hashlib class ContentValidator: def __init__(self, source_db, target_db): self.source_db source_db self.target_db target_db def generate_record_hash(self, record): sorted_keys sorted(record.keys()) hash_string |.join(f{k}{record[k]} for k in sorted_keys) return hashlib.md5(hash_string.encode()).hexdigest() def get_sample_data(self, db, table_name, sample_size100): result db.execute( fSELECT * FROM {table_name} ORDER BY id LIMIT %s, (sample_size,) ) columns [desc[0] for desc in result.description] return [dict(zip(columns, row)) for row in result.fetchall()] def validate_sample(self, table_name, sample_size100): source_sample self.get_sample_data(self.source_db, table_name, sample_size) target_sample self.get_sample_data(self.target_db, table_name, sample_size) mismatches [] for source_record, target_record in zip(source_sample, target_sample): source_hash self.generate_record_hash(source_record) target_hash self.generate_record_hash(target_record) if source_hash ! target_hash: mismatches.append({ id: source_record.get(id), source: source_record, target: target_record }) return { table: table_name, sample_size: sample_size, mismatches: mismatches, is_valid: len(mismatches) 0 } def validate_full(self, table_name): source_count self.source_db.execute(fSELECT COUNT(*) FROM {table_name}).fetchone()[0] target_count self.target_db.execute(fSELECT COUNT(*) FROM {table_name}).fetchone()[0] if source_count ! target_count: return { table: table_name, is_valid: False, message: fCount mismatch: source{source_count}, target{target_count} } batch_size 1000 mismatches [] for offset in range(0, source_count, batch_size): source_batch self.source_db.execute( fSELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s, (batch_size, offset) ).fetchall() target_batch self.target_db.execute( fSELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s, (batch_size, offset) ).fetchall() source_columns [desc[0] for desc in source_batch.description] target_columns [desc[0] for desc in target_batch.description] for source_row, target_row in zip(source_batch.fetchall(), target_batch.fetchall()): source_record dict(zip(source_columns, source_row)) target_record dict(zip(target_columns, target_row)) source_hash self.generate_record_hash(source_record) target_hash self.generate_record_hash(target_record) if source_hash ! target_hash: mismatches.append({ id: source_record.get(id), source: source_record, target: target_record }) return { table: table_name, total_records: source_count, mismatches: mismatches, is_valid: len(mismatches) 0 }4.4 结构校验结构校验代码示例class StructureValidator: def __init__(self, source_db, target_db): self.source_db source_db self.target_db target_db def get_table_structure(self, db, table_name): result db.execute(fDESCRIBE {table_name}) columns [] for row in result.fetchall(): columns.append({ name: row[0], type: row[1], null: row[2], key: row[3], default: row[4], extra: row[5] }) return columns def validate(self, table_name): source_structure self.get_table_structure(self.source_db, table_name) target_structure self.get_table_structure(self.target_db, table_name) source_columns {col[name]: col for col in source_structure} target_columns {col[name]: col for col in target_structure} missing_columns [name for name in source_columns if name not in target_columns] extra_columns [name for name in target_columns if name not in source_columns] type_mismatches [] for name in source_columns: if name in target_columns: source_type source_columns[name][type] target_type target_columns[name][type] if source_type ! target_type: type_mismatches.append({ column: name, source_type: source_type, target_type: target_type }) return { table: table_name, missing_columns: missing_columns, extra_columns: extra_columns, type_mismatches: type_mismatches, is_valid: not missing_columns and not extra_columns and not type_mismatches }五、一致性验证流程5.1 验证流程5.2 验证报告验证报告示例{ migration_id: migration_001, validation_time: 2024-01-15T10:00:00Z, tables: [ { name: users, count_validation: { source_count: 10000, target_count: 10000, is_valid: true }, content_validation: { sample_size: 100, mismatches: 0, is_valid: true }, structure_validation: { missing_columns: [], extra_columns: [], type_mismatches: [], is_valid: true }, overall_status: PASS } ], overall_status: PASS, total_tables: 1, passed_tables: 1, failed_tables: 0 }六、常见问题6.1 数据丢失现象目标数据少于源数据解决方案方案说明事务保障使用事务确保原子性校验点设置校验点支持断点续传日志记录记录每批数据处理日志6.2 数据重复现象目标数据存在重复记录解决方案方案说明唯一约束设置唯一约束幂等性设计实现幂等写入去重处理迁移前去重6.3 数据不一致现象源数据和目标数据不一致解决方案方案说明实时同步使用增量同步全量对比迁移后全量对比数据修复根据对比结果修复6.4 完整性约束违反现象外键约束或唯一约束违反解决方案方案说明顺序迁移按依赖顺序迁移延迟约束迁移完成后启用约束数据清理清理不符合约束的数据