Python脚本如何自定义分片同步批次大小

wen python案例 2026-06-26 260

本文目录导读：

基础分片同步架构
动态批次大小调整
数据库分片同步示例
文件分片同步
使用配置管理批次大小
最佳实践建议

在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')