本文目录导读：
我来介绍几种批量同步数据变更的Python方案：
基于SQL的批量同步
import pandas as pd
from sqlalchemy import create_engine
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DataSyncSQL:
def __init__(self, source_db, target_db):
self.source_engine = create_engine(source_db)
self.target_engine = create_engine(target_db)
def sync_by_timestamp(self, table_name, timestamp_column, batch_size=1000):
"""基于时间戳的增量同步"""
try:
# 获取目标表最大时间戳
query = f"SELECT MAX({timestamp_column}) FROM {table_name}"
last_sync = pd.read_sql(query, self.target_engine).iloc[0, 0]
if last_sync is None:
last_sync = '1970-01-01'
# 分批读取源数据
offset = 0
while True:
query = f"""
SELECT * FROM {table_name}
WHERE {timestamp_column} > '{last_sync}'
ORDER BY {timestamp_column}
LIMIT {batch_size} OFFSET {offset}
"""
df = pd.read_sql(query, self.source_engine)
if df.empty:
break
# 写入目标数据库
df.to_sql(table_name, self.target_engine,
if_exists='append', index=False)
offset += batch_size
logger.info(f"已同步 {offset} 条记录")
except Exception as e:
logger.error(f"同步失败: {e}")
raise
# 使用示例
sync = DataSyncSQL('mysql://user:pass@source_host/db',
'mysql://user:pass@target_host/db')
sync.sync_by_timestamp('orders', 'update_time', batch_size=5000)
基于变更捕获(CDC)的同步
import hashlib
import json
from datetime import datetime
class CDCDataSync:
def __init__(self, source_conn, target_conn):
self.source = source_conn
self.target = target_conn
self.sync_meta_table = 'sync_metadata'
def calculate_row_hash(self, row):
"""计算行哈希值用于变更检测"""
row_str = json.dumps(row, sort_keys=True, default=str)
return hashlib.md5(row_str.encode()).hexdigest()
def detect_changes(self, table_name, primary_keys, batch_size=1000):
"""检测并同步变更"""
source_rows = self.get_all_source_rows(table_name, batch_size)
target_hashes = self.get_target_hashes(table_name, primary_keys)
changes = {
'insert': [],
'update': [],
'delete': []
}
for row in source_rows:
pk = tuple(row[k] for k in primary_keys)
current_hash = self.calculate_row_hash(row)
if pk not in target_hashes:
changes['insert'].append(row)
elif target_hashes[pk] != current_hash:
changes['update'].append(row)
# 检测删除
source_pks = set(tuple(row[k] for k in primary_keys)
for row in source_rows)
for pk in target_hashes:
if pk not in source_pks:
changes['delete'].append(pk)
return changes
def apply_changes(self, table_name, changes, primary_keys):
"""应用变更到目标表"""
with self.target.cursor() as cursor:
# 批量插入
if changes['insert']:
insert_sql = self.build_insert_sql(table_name,
changes['insert'][0].keys())
cursor.executemany(insert_sql,
[list(row.values()) for row in changes['insert']])
# 批量更新
if changes['update']:
for row in changes['update']:
update_sql = self.build_update_sql(table_name,
row.keys(), primary_keys)
cursor.execute(update_sql, list(row.values()) +
[row[k] for k in primary_keys])
# 批量删除
if changes['delete']:
delete_sql = f"DELETE FROM {table_name} WHERE " + \
" AND ".join([f"{k} = %s" for k in primary_keys])
cursor.executemany(delete_sql, changes['delete'])
self.target.commit()
基于文件的批量同步
import csv
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
class FileBasedSync:
def __init__(self, source_files, target_dir, chunk_size=1000):
self.source_files = source_files
self.target_dir = target_dir
self.chunk_size = chunk_size
def process_chunk(self, chunk, file_index, chunk_index):
"""处理单个数据块"""
output_file = os.path.join(
self.target_dir,
f"chunk_{file_index}_{chunk_index}.csv"
)
with open(output_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(chunk[0].keys()) # 写入header
for row in chunk:
writer.writerow(row.values())
return output_file
def sync_files(self, max_workers=4):
"""并行同步多个文件"""
all_files = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = []
chunk_index = 0
for file_index, file_path in enumerate(self.source_files):
with open(file_path, 'r') as f:
reader = csv.DictReader(f)
chunk = []
for row in reader:
chunk.append(row)
if len(chunk) >= self.chunk_size:
future = executor.submit(
self.process_chunk,
chunk.copy(),
file_index,
chunk_index
)
futures.append(future)
chunk = []
chunk_index += 1
# 处理剩余数据
if chunk:
future = executor.submit(
self.process_chunk,
chunk,
file_index,
chunk_index
)
futures.append(future)
# 收集结果
for future in as_completed(futures):
try:
result = future.result()
all_files.append(result)
except Exception as e:
print(f"处理失败: {e}")
return all_files
企业级批量同步方案
from typing import List, Dict, Any
import asyncio
import aiohttp
import aiomysql
class EnterpriseDataSync:
def __init__(self, config: Dict):
self.config = config
self.batch_size = config.get('batch_size', 1000)
self.retry_count = config.get('retry_count', 3)
self.sync_strategy = config.get('sync_strategy', 'incremental')
async def sync_table(self, table_name: str,
where_condition: str = None):
"""异步同步单个表"""
try:
# 1. 获取源数据
source_data = await self.fetch_source_data(
table_name, where_condition
)
# 2. 数据转换和验证
validated_data = await self.validate_data(source_data)
# 3. 批量写入目标
await self.batch_write_target(validated_data, table_name)
# 4. 记录同步日志
await self.log_sync_result(table_name, len(validated_data))
except Exception as e:
await self.handle_sync_error(table_name, e)
async def batch_sync_all_tables(self, tables: List[str]):
"""批量同步多个表"""
tasks = []
for table in tables:
task = asyncio.create_task(
self.sync_table(table)
)
tasks.append(task)
# 等待所有同步任务完成
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
successful = []
failed = []
for table, result in zip(tables, results):
if isinstance(result, Exception):
failed.append((table, str(result)))
else:
successful.append(table)
return {
'successful': successful,
'failed': failed,
'timestamp': datetime.now().isoformat()
}
async def monitor_sync_progress(self):
"""监控同步进度"""
while True:
status = await self.get_sync_status()
print(f"同步进度: {status['progress']}%")
if status['completed']:
break
await asyncio.sleep(5)
实用同步工具类
import json
import dask.dataframe as dd
from typing import Optional
class DataSyncToolkit:
def __init__(self):
self.sync_config = {}
def compare_and_sync(self, source_df, target_df, keys):
"""比较并同步数据框"""
# 找出新增记录
new_records = source_df.merge(
target_df,
on=keys,
how='left',
indicator=True
).query('_merge == "left_only"')
# 找出变更记录
merged = source_df.merge(
target_df,
on=keys,
suffixes=('_new', '_old'),
how='inner'
)
changed = merged[merged['_merge'] == 'both']
return {
'new': new_records,
'changed': changed
}
def large_dataset_sync(self, source_path: str,
target_path: str,
chunk_size: int = 100000):
"""大数据集同步（使用Dask）"""
# 读取大数据集
ddf = dd.read_csv(source_path, blocksize=chunk_size)
# 数据转换
# 添加同步标记
ddf['sync_timestamp'] = datetime.now()
ddf['sync_status'] = 'pending'
# 保存到目标
ddf.to_parquet(
target_path,
compression='snappy',
overwrite=True
)
return True
def incremental_sync_with_state(self, source_query_func,
sync_state_file: str = 'sync_state.json'):
"""带状态记录的增量同步"""
# 读取同步状态
try:
with open(sync_state_file, 'r') as f:
last_sync_state = json.load(f)
except FileNotFoundError:
last_sync_state = {'last_id': 0}
# 执行增量查询
new_data = source_query_func(
last_id=last_sync_state['last_id']
)
if not new_data.empty:
# 执行同步逻辑
max_id = new_data['id'].max()
# 更新同步状态
last_sync_state['last_id'] = max_id
last_sync_state['last_sync_time'] = datetime.now().isoformat()
with open(sync_state_file, 'w') as f:
json.dump(last_sync_state, f, indent=2)
return new_data
使用建议
-
选择合适的同步策略：
- 全量同步：数据量小，首次同步
- 增量同步：数据量大，定期同步
- 实时同步：需要即时一致性
-
错误处理：
- 实现重试机制（指数退避）
- 记录失败记录，单独处理
- 事务回滚保证数据一致性
-
性能优化：
- 使用批量操作（BATCH）
- 并行处理不依赖的数据
- 使用索引加速查询
-
监控告警：
- 记录同步延迟
- 监控数据一致性
- 异常告警通知
根据你的具体场景（数据量、实时性要求、数据库类型等）选择合适的方案。
