本文目录导读：
对于海量数据同步,Python分批执行是必须的，否则会导致内存溢出或数据库连接超时，以下是几种常用的分批策略和实现方案：
基于LIMIT/OFFSET的分页分批（最常用）
适用于有自增ID或连续主键的表。
import time
from sqlalchemy import create_engine, text
def sync_data_in_batches(source_conn, target_conn, table_name, batch_size=10000):
"""
分批同步数据
"""
offset = 0
total_synced = 0
while True:
# 分批查询源数据库
query = f"""
SELECT * FROM {table_name}
ORDER BY id
LIMIT {batch_size} OFFSET {offset}
"""
with source_conn.connect() as conn:
result = conn.execute(text(query))
rows = result.fetchall()
if not rows:
break
# 批量插入目标数据库
with target_conn.connect() as conn:
for row in rows:
insert_query = f"""
INSERT INTO {table_name}
VALUES ({','.join(['?' for _ in row])})
ON CONFLICT(id) DO UPDATE SET ...
"""
conn.execute(text(insert_query), row)
conn.commit()
total_synced += len(rows)
offset += batch_size
print(f"已同步: {total_synced} 条记录")
time.sleep(0.1) # 避免数据库压力过大
return total_synced
基于游标的分批（更高效）
适用于大数据量,避免OFFSET的性能问题。
def sync_data_with_cursor(source_conn, target_conn, table_name, batch_size=10000):
"""
使用游标分批同步
"""
last_id = 0
total_synced = 0
while True:
# 基于最后一次ID进行分批
query = f"""
SELECT * FROM {table_name}
WHERE id > {last_id}
ORDER BY id
LIMIT {batch_size}
"""
with source_conn.connect() as conn:
result = conn.execute(text(query))
rows = result.fetchall()
if not rows:
break
# 准备批量插入
batch_data = []
for row in rows:
batch_data.append(dict(row._mapping))
# 批量写入目标库
with target_conn.connect() as conn:
# 使用executemany进行批量插入
conn.execute(
text(f"INSERT INTO {table_name} VALUES :data"),
{"data": batch_data}
)
conn.commit()
# 更新最后ID
last_id = rows[-1][0]
total_synced += len(rows)
print(f"已同步: {total_synced} 条记录, 当前ID: {last_id}")
return total_synced
基于时间窗口的分批
适用于按时间分区的数据同步。
from datetime import datetime, timedelta
def sync_data_by_time_window(source_conn, target_conn, table_name,
start_time, end_time, window_days=1):
"""
按时间窗口分批同步
"""
current_start = start_time
while current_start < end_time:
current_end = min(current_start + timedelta(days=window_days), end_time)
query = f"""
SELECT * FROM {table_name}
WHERE created_at >= '{current_start}'
AND created_at < '{current_end}'
ORDER BY created_at
"""
with source_conn.connect() as conn:
result = conn.execute(text(query))
rows = result.fetchall()
if rows:
# 批量插入
with target_conn.connect() as conn:
# 使用pandas进行高效批量插入
import pandas as pd
df = pd.DataFrame(rows)
df.to_sql(table_name, conn, if_exists='append',
index=False, chunksize=5000)
print(f"同步时间窗口: {current_start} - {current_end}, 记录数: {len(rows)}")
current_start = current_end
完整的同步框架（带断点续传）
import json
import os
from datetime import datetime
class DataSyncManager:
def __init__(self, config_file='sync_config.json'):
self.config = self.load_config(config_file)
self.checkpoint_file = 'sync_checkpoint.json'
def load_config(self, config_file):
"""加载配置文件"""
with open(config_file, 'r') as f:
return json.load(f)
def save_checkpoint(self, table_name, last_id, total_synced):
"""保存同步检查点"""
checkpoint = {}
if os.path.exists(self.checkpoint_file):
with open(self.checkpoint_file, 'r') as f:
checkpoint = json.load(f)
checkpoint[table_name] = {
'last_id': last_id,
'total_synced': total_synced,
'timestamp': str(datetime.now())
}
with open(self.checkpoint_file, 'w') as f:
json.dump(checkpoint, f, indent=2)
def load_checkpoint(self, table_name):
"""加载同步检查点"""
if not os.path.exists(self.checkpoint_file):
return None
with open(self.checkpoint_file, 'r') as f:
checkpoint = json.load(f)
return checkpoint.get(table_name)
def sync_table(self, source_config, target_config, table_name,
batch_size=5000, resume=True):
"""
执行表同步，支持断点续传
"""
# 连接数据库
source_conn = self.get_connection(source_config)
target_conn = self.get_connection(target_config)
try:
# 检查断点
last_id = 0
if resume:
checkpoint = self.load_checkpoint(table_name)
if checkpoint:
last_id = checkpoint['last_id']
print(f"从断点恢复: last_id={last_id}")
total_synced = 0
retry_count = 0
max_retries = 3
while True:
try:
# 分批查询
query = f"""
SELECT * FROM {table_name}
WHERE id > {last_id}
ORDER BY id
LIMIT {batch_size}
"""
with source_conn.connect() as conn:
result = conn.execute(text(query))
rows = result.fetchall()
if not rows:
break
# 批量插入
with target_conn.connect() as conn:
for row in rows:
# 构建插入语句
columns = list(row._mapping.keys())
values = list(row._mapping.values())
insert_sql = f"""
INSERT INTO {table_name} ({','.join(columns)})
VALUES ({','.join(['?' for _ in values])})
ON CONFLICT(id) DO UPDATE SET ...
"""
conn.execute(text(insert_sql), values)
conn.commit()
# 更新进度
last_id = rows[-1][0]
total_synced += len(rows)
retry_count = 0
# 保存检查点
self.save_checkpoint(table_name, last_id, total_synced)
print(f"表 {table_name}: 已同步 {total_synced} 条记录")
except Exception as e:
retry_count += 1
if retry_count > max_retries:
raise
print(f"同步出错，重试 {retry_count}/{max_retries}: {e}")
time.sleep(5)
print(f"表 {table_name} 同步完成，共 {total_synced} 条记录")
return total_synced
finally:
source_conn.close()
target_conn.close()
def get_connection(self, config):
"""获取数据库连接"""
from sqlalchemy import create_engine
conn_str = f"{config['type']}://{config['user']}:{config['password']}@" \
f"{config['host']}:{config['port']}/{config['database']}"
return create_engine(conn_str)
并行分批同步（多线程/多进程）
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
class ParallelDataSync:
def __init__(self, max_workers=4):
self.max_workers = max_workers
self.lock = threading.Lock()
def sync_table_parallel(self, source_config, target_config, table_name):
"""
并行同步单表的不同分区
"""
# 获取表的总ID范围
source_conn = self.get_connection(source_config)
with source_conn.connect() as conn:
result = conn.execute(text(f"SELECT MIN(id), MAX(id) FROM {table_name}"))
min_id, max_id = result.fetchone()
# 计算分区
total_records = max_id - min_id + 1
chunk_size = total_records // self.max_workers
futures = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
for i in range(self.max_workers):
start_id = min_id + i * chunk_size
end_id = start_id + chunk_size if i < self.max_workers - 1 else max_id
future = executor.submit(
self.sync_range,
source_config, target_config, table_name,
start_id, end_id
)
futures.append(future)
total_synced = 0
for future in as_completed(futures):
total_synced += future.result()
return total_synced
性能优化建议
def optimized_batch_insert(target_conn, table_name, rows, batch_size=5000):
"""
优化批量插入性能
"""
import psycopg2.extras
# 使用copy_from（PostgreSQL）
with target_conn.cursor() as cursor:
# 准备CSV数据
import io
csv_buffer = io.StringIO()
writer = csv.writer(csv_buffer)
for row in rows:
writer.writerow(row)
csv_buffer.seek(0)
cursor.copy_from(csv_buffer, table_name, sep=',', null='')
target_conn.commit()
# 或使用批量插入优化
with target_conn.cursor() as cursor:
psycopg2.extras.execute_values(
cursor,
f"INSERT INTO {table_name} VALUES %s",
rows,
template=None,
page_size=batch_size
)
target_conn.commit()
关键配置示例
{
"source": {
"type": "postgresql",
"host": "source_host",
"port": 5432,
"database": "source_db",
"user": "user",
"password": "password"
},
"target": {
"type": "postgresql",
"host": "target_host",
"port": 5432,
"database": "target_db",
"user": "user",
"password": "password"
},
"sync_settings": {
"batch_size": 10000,
"parallel_workers": 4,
"retry_times": 3,
"checkpoint_enabled": true
}
}
注意事项
- 事务管理：每批处理完及时提交，避免长事务
- 错误处理：实现重试机制和错误日志
- 性能监控：记录每批处理时间和速度
- 资源控制：监控内存使用，避免OOM
- 数据一致性：使用事务确保原子性
- 索引优化：大批量插入前暂时禁用索引，插入后重建
选择哪种策略取决于你的数据特点和业务需求,通常建议从简单的LIMIT/OFFSET实现开始，根据实际性能进行优化。
