本文介绍通过优化 sql 插入语句、正确配置 schema.ini 文件及一次性全量导入的方式，彻底规避 ms access “file sharing lock count exceeded” 错误，适用于百万级数据导入场景。
本文介绍通过优化 sql 插入语句、正确配置 schema.ini 文件及一次性全量导入的方式，彻底规避 ms access “file sharing lock count exceeded” 错误，适用于百万级数据导入场景。
在使用 Python 向 MS Access 批量导入大量数据（如 150 万行）时，频繁分块 + 多次 INSERT 操作极易触发系统级限制——“File Sharing Lock Count Exceeded”（默认最大锁数为 9500）。根本原因在于：Access 的 Jet/ACE 引擎对每个事务中涉及的记录锁数量有硬性上限，而传统分块插入（尤其是基于临时 CSV + SELECT INTO 的循环方式）会持续累积未释放的锁，即使调小 chunk_size 或调整 commit_interval 也难以突破该瓶颈。
✅ 正确解法不是“更细地切块”，而是绕过锁竞争机制本身：利用 Access 原生支持的 Text ISAM 驱动直连导入能力，将整张 DataFrame 一次性导出为规范 CSV，并通过 INSERT ... SELECT FROM [Text;...] 语句交由 Access 引擎内部高效加载——此过程不经过 ODBC 行级操作，几乎不产生用户可见锁，从而彻底规避锁计数超限问题。
关键实践要点
-
*显式声明字段，禁用 `SELECT
** Access 对SELECT *` 的列映射容错性差，且易因类型推断失败导致锁异常。必须显式列出目标表与源 CSV 的对应字段：df_cols = df_adjusted.columns.tolist() sql = ( f"INSERT INTO [{table_name}] ({', '.join(df_cols)}) " f"SELECT {', '.join(df_cols)} " f"FROM [text;database={os.path.dirname(temp_file_path)}].[{os.path.basename(temp_file_path)}]" ) -
严格匹配文件格式并配置
schema.iniDataFrame.to_csv()默认生成逗号分隔、带表头的 CSV；但原始代码中 SQL 使用FMT=TabDelimited，造成格式错配，引发解析失败与隐式锁重试。务必统一为 CSV 格式，并在 CSV 所在目录创建schema.ini文件，明确告知 Access 如何解析：
Web Accessibility (WCAG 2 AA))下载将 WCAG 2 AA 无障碍标准应用于设计系统、UI 组件和网页——通过可运行的验证器进行对比度计算，采用设计 token 模式使颜色天生可访问，设定排版和焦点状态规则，并提供语义检查清单。适用于构建或审查 UI。
[your_data.csv] Format=CSVDelimited ColNameHeader=True MaxScanRows=500
⚠️ 注意：
MaxScanRows必须足够大（建议 ≥500），否则 Access 仅扫描前几行就推断列类型，遇到含空值或混合数值格式的列时会错误设为Integer或Text，后续导入失败或数据截断，间接引发重试和锁堆积。 单次提交，避免中间
commit()干扰
全量导入语句执行期间，Access 内部以最小化锁粒度完成加载，无需外部事务控制。整个操作应封装为单次cursor.execute(sql)+ 单次conn.commit()，杜绝循环中频繁提交带来的资源残留。
完整推荐代码片段
def bulk_insert_to_access_optimized(file_path, table_name, temp_file_path=None, connect_str=None):
"""高性能无锁批量导入：一次性全量 CSV → Access"""
if not connect_str:
raise ValueError("Database connection string is required")
# 1. 加载并清洗数据
df = pd.read_csv(file_path, delimiter="\t", encoding="utf-8")
column_specs = get_column_specs(table_name)
df_adjusted = adjust_data_to_specs(df, column_specs)
# 2. 导出为标准 CSV（带表头）
if not temp_file_path:
temp_file_path = os.path.join(os.path.dirname(file_path), "temp_import.csv")
df_adjusted.to_csv(temp_file_path, index=False)
# 3. 自动生成 schema.ini（同目录）
schema_ini_path = os.path.join(os.path.dirname(temp_file_path), "schema.ini")
with open(schema_ini_path, "w", encoding="utf-8") as f:
f.write(f"""[{os.path.basename(temp_file_path)}]
Format=CSVDelimited
ColNameHeader=True
MaxScanRows=500
""")
# 4. 构建安全 SQL（显式字段 + 路径转义）
df_cols = df_adjusted.columns.tolist()
db_dir = os.path.dirname(temp_file_path)
csv_name = os.path.basename(temp_file_path)
sql = (
f"INSERT INTO [{table_name}] ({', '.join([f'[{c}]' for c in df_cols])}) "
f"SELECT {', '.join([f'[{c}]' for c in df_cols])} "
f"FROM [text;database={db_dir}].[{csv_name}]"
)
# 5. 执行导入
conn = pyodbc.connect(connect_str)
cursor = conn.cursor()
try:
cursor.execute(sql)
conn.commit()
print(f"✅ Successfully imported {len(df_adjusted)} rows into [{table_name}]")
except pyodbc.Error as e:
conn.rollback()
raise RuntimeError(f"Access import failed: {e}")
finally:
cursor.close()
conn.close()
# 可选：清理临时文件
for f in [temp_file_path, schema_ini_path]:
if os.path.exists(f):
os.remove(f)
总结与注意事项
- ✅ 性能提升显著：实测 200–800 万行导入通常在 2–5 分钟内完成，远快于分块方案；
- ❌ 禁止混用格式：
.to_csv()→CSVDelimited；read_csv(..., sep='\t')→TabDelimited，二者必须严格一致； - ? 权限与路径要求：Access 进程需对 CSV 所在目录有读取权限；路径避免中文/空格，推荐使用短英文路径；
- ? 字段名兼容性：若列名含空格或特殊字符，SQL 中务必用方括号包裹（如
[Customer Name]），代码已自动处理； - ? 内存考量：虽不增加锁压力，但需确保 Python 进程内存足以容纳完整 DataFrame（150 万行 × 数十列通常
该方法本质是“借力 Access 自身引擎”，而非对抗其设计限制。在无法迁移至 SQL Server 等现代数据库的前提下，这是兼顾稳定性、性能与维护性的最优解。
