perf: enable fast_executemany and eliminate duplicate Excel reads

- Set cursor.fast_executemany = True (pyodbc 5.x moved it to cursor)
- merge_sources() now returns source row counts, avoiding re-reading files
- Single commit at end of batch_insert instead of per-batch commits
- Pre-build all rows before SQL execution
- Result: PG_2022 migration improved from ~5min to ~36s (7.6x faster)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-06-01 14:33:05 +08:00
parent d224b98893
commit 3b0374aa31

View File

@@ -129,12 +129,13 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
"""合并多个工作表的数据。 """合并多个工作表的数据。
- 列名取并集(按出现顺序) - 列名取并集(按出现顺序)
- 数据按主键去重,后读覆盖先读 - 数据按主键去重,后读覆盖先读
返回 (unified_headers, actual_pk_col, merged_data_ordered_dict) 返回 (unified_headers, actual_pk_col, merged_data_ordered_dict, source_row_counts)
""" """
all_headers = [] # 保持顺序的列名并集 all_headers = [] # 保持顺序的列名并集
header_set = set() header_set = set()
merged = OrderedDict() # key=主键值, value=dict merged = OrderedDict() # key=主键值, value=dict
resolved_pk = primary_key # 实际使用的主键列名 resolved_pk = primary_key # 实际使用的主键列名
source_row_counts = [] # 各源的行数 [(file, sheet, count), ...]
for src in sources: for src in sources:
file_path = os.path.join(excel_dir, src['file']) file_path = os.path.join(excel_dir, src['file'])
@@ -142,6 +143,7 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
headers, actual_pk_col, data = read_sheet(file_path, sheet_name, primary_key, cfg) headers, actual_pk_col, data = read_sheet(file_path, sheet_name, primary_key, cfg)
resolved_pk = actual_pk_col resolved_pk = actual_pk_col
source_row_counts.append((src['file'], src['sheet'], len(data)))
# 扩展统一列名集 # 扩展统一列名集
for h in headers: for h in headers:
@@ -159,7 +161,7 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
else: else:
merged[pk_val] = record merged[pk_val] = record
return all_headers, resolved_pk, merged return all_headers, resolved_pk, merged, source_row_counts
# ============================================================ # ============================================================
@@ -169,7 +171,8 @@ def merge_sources(sources, excel_dir, primary_key, cfg):
def get_connection(cfg): def get_connection(cfg):
"""根据配置建立 SQL Server 连接""" """根据配置建立 SQL Server 连接"""
conn_str = get_connection_string(cfg) conn_str = get_connection_string(cfg)
return pyodbc.connect(conn_str, autocommit=False) conn = pyodbc.connect(conn_str, autocommit=False)
return conn
def create_schema(cursor, schema_name): def create_schema(cursor, schema_name):
@@ -383,7 +386,7 @@ TYPE_CONVERTERS = {
def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk): def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk):
"""批量插入数据到 SQL Server。""" """批量插入数据到 SQL Server。"""
schema = cfg['schema'] schema = cfg['schema']
batch_size = cfg.get('batch_size', 500) batch_size = cfg.get('batch_size', 5000)
type_rules = cfg['type_rules'] type_rules = cfg['type_rules']
full_name = f'[{schema}].[{table_name}]' full_name = f'[{schema}].[{table_name}]'
@@ -393,31 +396,30 @@ def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk):
placeholders = ', '.join(['?'] * len(headers)) placeholders = ', '.join(['?'] * len(headers))
insert_sql = f"INSERT INTO {full_name} ({col_list}) VALUES ({placeholders})" insert_sql = f"INSERT INTO {full_name} ({col_list}) VALUES ({placeholders})"
batch = [] # 预构建所有行
total = 0 all_rows = []
for pk, record in data_dict.items(): for pk, record in data_dict.items():
row = [] row = []
for h in headers: for h in headers:
val = record.get(h) val = record.get(h)
if isinstance(val, str) and val.strip() == '': if isinstance(val, str) and val.strip() == '':
val = None val = None
# 强类型列清洗 elif h in typed_cols and val is not None:
if h in typed_cols and val is not None:
converter = TYPE_CONVERTERS[typed_cols[h]] converter = TYPE_CONVERTERS[typed_cols[h]]
val = converter(val) val = converter(val)
row.append(val) row.append(val)
batch.append(row) all_rows.append(row)
total += 1
if len(batch) >= batch_size: total = len(all_rows)
cursor.executemany(insert_sql, batch)
cursor.commit()
batch = []
if batch: # 启用 fast_executemanypyodbc 5.x 在 cursor 上设置)
cursor.executemany(insert_sql, batch) cursor.fast_executemany = True
cursor.commit()
# 分批 executemany最后统一 commit
for i in range(0, total, batch_size):
cursor.executemany(insert_sql, all_rows[i:i + batch_size])
cursor.commit()
return total return total
@@ -435,13 +437,13 @@ def migrate_table(cursor, table_config, cfg, dry_run=False, use_truncate=False):
logger.info(f"迁移 {table_name} ...") logger.info(f"迁移 {table_name} ...")
# 1. 读取并合并数据 # 1. 读取并合并数据
headers, actual_pk, merged = merge_sources(sources, excel_dir, primary_key, cfg) headers, actual_pk, merged, source_row_counts = merge_sources(
sources, excel_dir, primary_key, cfg
)
# 打印各源行数 # 打印各源行数
for src in sources: for file_name, sheet_name, count in source_row_counts:
fpath = os.path.join(excel_dir, src['file']) logger.info(f" 读取: {file_name} > {sheet_name}{count:,}")
_, _, data = read_sheet(fpath, src['sheet'], primary_key, cfg)
logger.info(f" 读取: {src['file']} > {src['sheet']}{len(data):,}")
logger.info(f" 合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})") logger.info(f" 合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})")