本文目录导读：
我来介绍几种批量同步数据变更的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）
- 并行处理不依赖的数据
- 使用索引加速查询
-
监控告警：
- 记录同步延迟
- 监控数据一致性
- 异常告警通知
根据你的具体场景（数据量、实时性要求、数据库类型等）选择合适的方案。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
