Python脚本如何拆分大表同步任务分片

wen python案例 2026-06-26 149

本文目录导读：

基于主键范围分片（推荐）
基于时间范围分片
并行执行方案
动态调整分片大小
断点续传支持
使用建议：

对于Python拆分大表同步任务分片,以下是几种常用方案：

基于主键范围分片（推荐）

import pymysql
from typing import List, Tuple
import time
class TableShardSplitter:
    def __init__(self, db_config, table_name, primary_key='id', shard_size=10000):
        self.db_config = db_config
        self.table_name = table_name
        self.primary_key = primary_key
        self.shard_size = shard_size
    def get_shards(self) -> List[Tuple[int, int]]:
        """获取分片范围"""
        conn = pymysql.connect(**self.db_config)
        cursor = conn.cursor()
        # 获取主键最小值和最大值
        cursor.execute(f"SELECT MIN({self.primary_key}), MAX({self.primary_key}) FROM {self.table_name}")
        min_id, max_id = cursor.fetchone()
        cursor.close()
        conn.close()
        if not min_id or not max_id:
            return []
        # 按固定大小分片
        shards = []
        current = min_id
        while current <= max_id:
            end = min(current + self.shard_size - 1, max_id)
            shards.append((current, end))
            current = end + 1
        return shards
# 使用示例
def sync_shard(table_name, start_id, end_id, db_config, target_config):
    """同步单个分片数据"""
    conn = pymysql.connect(**db_config)
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    # 读取分片数据
    cursor.execute(
        f"SELECT * FROM {table_name} WHERE id BETWEEN %s AND %s",
        (start_id, end_id)
    )
    records = cursor.fetchall()
    # 处理数据（示例：写入目标表）
    process_records(records, target_config)
    cursor.close()
    conn.close()
    print(f"Sync completed: id {start_id} - {end_id}, records: {len(records)}")
# 主执行函数
def main_sync():
    db_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'source_db'
    }
    target_config = {
        'host': 'target_host',
        'user': 'target_user',
        'password': 'target_pass',
        'database': 'target_db'
    }
    # 创建分片器
    splitter = TableShardSplitter(
        db_config=db_config,
        table_name='big_table',
        primary_key='id',
        shard_size=10000  # 每片1万条
    )
    # 获取分片
    shards = splitter.get_shards()
    print(f"Total shards: {len(shards)}")
    # 顺序执行（可以改为多线程并行）
    for start_id, end_id in shards:
        sync_shard('big_table', start_id, end_id, db_config, target_config)

基于时间范围分片

class TimeBasedSharder:
    def get_time_shards(self, table_name, time_column, start_date, end_date, interval_days=7):
        """按时间范围分片"""
        from datetime import datetime, timedelta
        shards = []
        current = start_date
        while current < end_date:
            next_date = min(current + timedelta(days=interval_days), end_date)
            shards.append((current.strftime('%Y-%m-%d'), next_date.strftime('%Y-%m-%d')))
            current = next_date
        return shards
    def sync_time_shard(self, table_name, start_date, end_date, db_config):
        """按时间范围同步"""
        conn = pymysql.connect(**db_config)
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute(f"""
            SELECT * FROM {table_name} 
            WHERE {self.time_column} BETWEEN %s AND %s
        """, (start_date, end_date))
        records = cursor.fetchall()
        # 处理数据...
        cursor.close()
        conn.close()

并行执行方案

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
class ParallelShardSync:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers
        self.lock = threading.Lock()
    def parallel_sync(self, shards, table_name, db_config, target_config):
        """并行执行分片同步"""
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            for shard in shards:
                future = executor.submit(
                    self.sync_shard_worker,
                    shard, table_name, db_config, target_config
                )
                futures.append(future)
            # 收集结果
            for future in as_completed(futures):
                try:
                    result = future.result()
                    print(f"Shard completed: {result}")
                except Exception as e:
                    print(f"Shard failed: {e}")
    def sync_shard_worker(self, shard, table_name, db_config, target_config):
        """工作线程函数"""
        try:
            # 创建独立数据库连接（避免线程冲突）
            local_conn = pymysql.connect(**db_config)
            # 执行同步逻辑
            sync_shard(table_name, shard[0], shard[1], db_config, target_config)
        finally:
            if local_conn:
                local_conn.close()

动态调整分片大小

class AdaptiveShardSplitter:
    def __init__(self, target_records_per_shard=10000, max_retries=3):
        self.target_records = target_records_per_shard
        self.max_retries = max_retries
    def adaptive_shard(self, db_config, table_name):
        """动态调整分片大小"""
        conn = pymysql.connect(**db_config)
        cursor = conn.cursor()
        # 获取表统计信息
        cursor.execute(f"""
            SELECT 
                COUNT(*) as total_records,
                CEIL(COUNT(*) / {self.target_records}) as estimated_shards
            FROM {table_name}
        """)
        stats = cursor.fetchone()
        total_records = stats[0]
        estimated_shards = stats[1]
        # 计算实际分片大小
        actual_shard_size = max(5000, total_records // estimated_shards)
        # 获取分片边界
        cursor.execute(f"""
            SELECT 
                id, 
                NTILE({estimated_shards}) OVER (ORDER BY id) as bucket
            FROM {table_name}
            WHERE MOD(id, 100) = 0
            GROUP BY bucket
        """)
        shard_boundaries = []
        for row in cursor.fetchall():
            if row[1] in [1, estimated_shards]:
                shard_boundaries.append(row[0])
        cursor.close()
        conn.close()
        return shard_boundaries

断点续传支持

class ResumeableShardSync:
    def __init__(self, progress_file='shard_progress.json'):
        self.progress_file = progress_file
        self.completed_shards = self.load_progress()
    def load_progress(self):
        """加载同步进度"""
        import json
        import os
        if os.path.exists(self.progress_file):
            with open(self.progress_file, 'r') as f:
                return set(tuple(shard) for shard in json.load(f))
        return set()
    def save_progress(self, shard):
        """保存同步进度"""
        import json
        self.completed_shards.add(shard)
        with open(self.progress_file, 'w') as f:
            json.dump(list(self.completed_shards), f)
    def sync_with_progress(self, all_shards, sync_func):
        """带断点续传的同步"""
        remaining_shards = [s for s in all_shards if s not in self.completed_shards]
        for shard in remaining_shards:
            try:
                sync_func(shard)
                self.save_progress(shard)
                print(f"Completed shard: {shard}")
            except Exception as e:
                print(f"Failed shard {shard}: {e}")
                # 记录失败的分片，后续重试
                raise