本文目录导读：
对于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
使用建议：
-
选择分片策略：
- 有自增ID：用ID范围分片
- 有时间字段：用时间分片
- 没有索引：考虑ROW_NUMBER()分页
-
性能优化：
- 使用连接池管理数据库连接
- 批量插入（每次500-1000条）
- 关闭自动提交，事务提交
-
异常处理：
- 断点续传机制
- 失败重试策略
- 监控和告警
根据你的具体场景选择合适的方案，生产环境建议使用分布式调度框架（如Apache Airflow）来管理这些同步任务。
