本文目录导读：
对于海量数据同步,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实现开始，根据实际性能进行优化。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
