本文目录导读：
在Python中自定义分片同步的批次大小,主要涉及以下几个方面：
基础分片同步架构
import threading
from queue import Queue
from typing import List, Any, Callable
class ShardSync:
def __init__(self, batch_size: int = 1000, max_workers: int = 4):
self.batch_size = batch_size
self.max_workers = max_workers
self.queue = Queue()
def sync_data(self, data_source: List[Any], sync_func: Callable):
"""
自定义批次大小的分片同步
"""
# 分片处理
shards = self._create_shards(data_source)
# 多线程同步
threads = []
for shard in shards:
thread = threading.Thread(
target=self._sync_shard,
args=(shard, sync_func)
)
threads.append(thread)
thread.start()
# 等待所有线程完成
for thread in threads:
thread.join()
def _create_shards(self, data: List[Any]) -> List[List[Any]]:
"""根据批次大小创建分片"""
shards = []
for i in range(0, len(data), self.batch_size):
shard = data[i:i + self.batch_size]
shards.append(shard)
return shards
def _sync_shard(self, shard: List[Any], sync_func: Callable):
"""同步单个分片"""
try:
result = sync_func(shard)
self.queue.put(('success', result))
except Exception as e:
self.queue.put(('error', str(e)))
动态批次大小调整
import time
from dataclasses import dataclass
@dataclass
class BatchConfig:
initial_size: int = 1000
min_size: int = 100
max_size: int = 10000
scale_factor: float = 1.5
class AdaptiveBatchSync:
def __init__(self, config: BatchConfig = None):
self.config = config or BatchConfig()
self.current_batch_size = self.config.initial_size
self.performance_metrics = []
def adaptive_sync(self, data: List[Any], sync_func: Callable) -> dict:
"""
自适应调整批次大小的同步
"""
total_start = time.time()
processed = 0
errors = 0
while processed < len(data):
# 动态调整批次大小
batch_size = self._calculate_batch_size()
# 获取当前批次数据
batch = data[processed:processed + batch_size]
# 同步并计时
start = time.time()
try:
sync_func(batch)
elapsed = time.time() - start
self._record_performance(batch_size, elapsed, True)
except Exception as e:
elapsed = time.time() - start
self._record_performance(batch_size, elapsed, False)
errors += 1
# 失败时减小批次大小
self.current_batch_size = max(
self.config.min_size,
int(self.current_batch_size / 2)
)
continue
processed += len(batch)
return {
'total_time': time.time() - total_start,
'processed': processed,
'errors': errors,
'final_batch_size': self.current_batch_size
}
def _calculate_batch_size(self) -> int:
"""计算最优批次大小"""
if not self.performance_metrics:
return self.current_batch_size
# 基于最近性能数据调整
recent_metrics = self.performance_metrics[-10:]
avg_throughput = sum(
m['records_per_second'] for m in recent_metrics
) / len(recent_metrics)
# 根据吞吐量调整
if avg_throughput > 1000: # 吞吐量高，增加批次
self.current_batch_size = min(
self.config.max_size,
int(self.current_batch_size * self.config.scale_factor)
)
else: # 吞吐量低，减小批次
self.current_batch_size = max(
self.config.min_size,
int(self.current_batch_size / self.config.scale_factor)
)
return self.current_batch_size
def _record_performance(self, batch_size: int, elapsed: float, success: bool):
"""记录性能指标"""
records_per_second = batch_size / elapsed if elapsed > 0 else 0
self.performance_metrics.append({
'batch_size': batch_size,
'elapsed': elapsed,
'success': success,
'records_per_second': records_per_second
})
数据库分片同步示例
import sqlite3
from contextlib import contextmanager
class DatabaseShardSync:
def __init__(self, db_path: str, batch_size: int = 500):
self.db_path = db_path
self.batch_size = batch_size
@contextmanager
def get_connection(self):
conn = sqlite3.connect(self.db_path)
try:
yield conn
finally:
conn.close()
def sync_table_by_batch(self, source_table: str, target_table: str,
where_condition: str = None):
"""
按批次同步数据库表
"""
with self.get_connection() as conn:
cursor = conn.cursor()
# 获取总记录数
count_sql = f"SELECT COUNT(*) FROM {source_table}"
if where_condition:
count_sql += f" WHERE {where_condition}"
cursor.execute(count_sql)
total_records = cursor.fetchone()[0]
# 分批次同步
offset = 0
while offset < total_records:
# 读取当前批次数据
select_sql = f"""
SELECT * FROM {source_table}
WHERE rowid > {offset}
LIMIT {self.batch_size}
"""
if where_condition:
select_sql = select_sql.replace(
f"WHERE rowid > {offset}",
f"WHERE {where_condition} AND rowid > {offset}"
)
cursor.execute(select_sql)
batch_data = cursor.fetchall()
if not batch_data:
break
# 批量插入到目标表
self._bulk_insert(conn, target_table, batch_data)
conn.commit()
offset += len(batch_data)
def _bulk_insert(self, conn, table: str, data: List[tuple]):
"""批量插入数据"""
if not data:
return
# 构建批量插入SQL
placeholders = ','.join(['?' * len(data[0])])
insert_sql = f"INSERT INTO {table} VALUES ({placeholders})"
conn.executemany(insert_sql, data)
文件分片同步
import os
from pathlib import Path
class FileShardSync:
def __init__(self, batch_size_mb: int = 10):
self.batch_size_bytes = batch_size_mb * 1024 * 1024
def sync_large_file(self, source_path: str, target_path: str,
callback: Callable[[int, int], None] = None):
"""
大文件分片同步
"""
file_size = os.path.getsize(source_path)
bytes_synced = 0
with open(source_path, 'rb') as src, open(target_path, 'wb') as dst:
while bytes_synced < file_size:
# 读取当前批次
remaining = file_size - bytes_synced
read_size = min(self.batch_size_bytes, remaining)
chunk = src.read(read_size)
if not chunk:
break
# 写入目标文件
dst.write(chunk)
bytes_synced += len(chunk)
# 回调进度
if callback:
callback(bytes_synced, file_size)
return bytes_synced
使用配置管理批次大小
import yaml
from typing import Optional
class BatchSizeManager:
def __init__(self, config_file: Optional[str] = None):
self.config = self._load_config(config_file)
self.batch_sizes = {}
def _load_config(self, config_file: Optional[str] = None):
"""加载配置文件"""
default_config = {
'batch_size': {
'default': 1000,
'min': 100,
'max': 10000,
'database': 500,
'file': 10485760, # 10MB
'api': 100
}
}
if config_file and os.path.exists(config_file):
with open(config_file, 'r') as f:
user_config = yaml.safe_load(f)
default_config.update(user_config)
return default_config
def get_batch_size(self, sync_type: str = 'default') -> int:
"""获取指定类型的批次大小"""
return self.config['batch_size'].get(
sync_type,
self.config['batch_size']['default']
)
def set_batch_size(self, sync_type: str, size: int):
"""动态设置批次大小"""
min_size = self.config['batch_size']['min']
max_size = self.config['batch_size']['max']
# 限制在合理范围内
size = max(min_size, min(max_size, size))
self.batch_sizes[sync_type] = size
# 使用示例
if __name__ == "__main__":
# 基础使用
sync = ShardSync(batch_size=500)
# 自适应批次同步
adaptive_sync = AdaptiveBatchSync()
# 数据库同步
db_sync = DatabaseShardSync('data.db', batch_size=1000)
# 使用配置管理
config_manager = BatchSizeManager('sync_config.yaml')
db_batch_size = config_manager.get_batch_size('database')
最佳实践建议
- 根据数据特性调整：不同类型的数据（文本、二进制、大对象）需要不同的批次大小
- 监控性能指标：实时监控同步速度和错误率，动态调整批次大小
- 设置合理的范围：限制批次大小在合理范围内，避免过大或过小
- 错误重试机制：批次失败时，减小批次大小重试
- 日志记录：记录每次同步的详细信息，便于问题排查
通过以上方法,你可以灵活地自定义分片同步的批次大小，根据实际需求进行优化。
