问题背景
在数据处理工作中，经常需要应对大量 CSV 文件的整合任务。这些文件通常具有相同的结构，需合并到一个统一的数据源中进行查询和分析。SQLite 是一款轻量级嵌入式数据库，特别适合此类场景，尤其是在数据规模适中、无需复杂数据库服务器的情况下。
解决方案：将多个 CSV 文件导入 SQLite 并构建主表
下面是一个完整的操作流程，展示如何借助 Python 和 SQLite 来处理大批量的 CSV 文件，并建立一个汇总所有数据的主表。
1. 环境准备
首先确认是否已安装必要的 Python 模块：pandas 和 sqlite3。如未安装，可通过 pip 命令安装：
pip install pandas sqlite3
2. 编写执行代码
import pandas as pd import sqlite3 import os import re设置 CSV 文件所在路径
csv_directory = r"path/to/your/csv/files" # 将 "path/to/your/csv/files" 替换为实际路径
立即学习“Python免费学习笔记（深入）”；
连接或创建 SQLite 数据库
conn = sqlite3.connect("db.db") cursor = conn.cursor()
获取目录下所有 CSV 文件
csv_files = [file for file in os.listdir(csv_directory) if file.endswith(".csv")]
定义主表名称
master_bus = "master_table"
清理主表名中的非法字符
master_bus = re.sub(r'\W ', '', master_bus) # 移除非字母数字字符
遍历每个 CSV 文件并导入数据库
for csv_file in csv_files: table_name = os.path.splitext(csv_file)[0] csv_path = os.path.join(csv_directory, csv_file)
# 使用 pandas 读取 CSV 文件 df = pd.read_csv(csv_path) # 添加记录来源文件名的新列 df['filename'] = os.path.basename(csv_file) # 将 DataFrame 写入数据库，表名为文件名 df.to_sql(table_name, conn, index=False, if_exists='replace') # 若表存在则替换获取当前数据库中所有表名
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") tables = cursor.fetchall() tables = [table[0] for table in tables]
创建主表
先检查是否存在同名主表，如有则删除
cursor.execute(f"DROP TABLE IF EXISTS {master_bus}")
根据第一个子表结构创建空主表
if tables: # 确保有子表存在 cursor.execute(f"CREATE TABLE {master_bus} AS SELECT * FROM {tables[0]} WHERE 0;")
# 将所有子表数据插入主表 for table in tables: cursor.execute(f"INSERT INTO {master_bus} SELECT * FROM {table};")else: print("没有发现可用子表，无法创建主表。")
提交事务并关闭连接
conn.commit() conn.close()
print("数据处理完成！")
3. 代码说明
- 模块引入: 导入 pandas 用于加载 CSV 数据，sqlite3 用于数据库交互，os 用于文件操作，re 用于清理表名。
- 路径配置与数据库连接: 设定 CSV 文件存储路径，并建立与 SQLite 的连接。
- 获取文件列表: 利用 os.listdir 函数列出目标目录下的所有 CSV 文件。
- 定义主表名: 设置一个统一的主表名，便于后续合并数据。
- 逐个处理 CSV 文件:
- 用 pandas 读取文件内容生成 DataFrame。
- 新增一列记录文件名，方便后期识别数据来源。
- 将数据写入 SQLite 表，表名基于原始 CSV 文件名。if_exists='replace' 参数确保不会因重复建表报错。
- 构建主表:
- 检查主表是否已存在，若存在则先清除。
- 从第一个子表复制结构创建一个空的主表。
- 循环遍历所有子表，把数据合并进主表。
- 提交变更并断开连接: 执行 commit 提交所有更改，然后关闭数据库连接。
4. 处理 OperationalError: near "-": syntax error 错误
该错误多由表名含特殊字符（如 -）导致。解决办法是通过正则表达式移除非法字符：
master_bus = re.sub(r'\W ', '', master_bus)此行代码利用正则表达式匹配所有非字母数字字符并替换为空，以保证表名合法。
5. 注意事项及优化建议
异常处理: 实际应用时应加入 try...except 结构来捕获可能发生的异常，比如文件找不到、读取失败等。
-
性能调优: 面对超大体积的 CSV 文件，可采用分批读取方式，并配合 chunksize 参数：
for chunk in pd.read_csv(csv_path, chunksize=10000): chunk.to_sql(table_name, conn, index=False, if_exists='append')
此方式每次处理 10000 行数据，避免内存溢出。
-
索引设置: 如需频繁查询主表，建议添加索引提升效率。例如：
cursor.execute(f"CREATE INDEX idx_filename ON {master_bus}(filename);") 字段类型控制: 默认情况下，pandas 会自动推断字段类型。如需精确控制，可在 to_sql 方法中通过 dtype 参数指定各列类型。
-
事务管理: 对大规模数据写入操作，启用事务可显著提升性能。示例：
conn.execute("BEGIN TRANSACTION") try:执行批量写入操作
conn.commit()except: conn.rollback() raise
6. 总结
本文详细讲解了如何结合 Python 与 SQLite 来处理大量 CSV 文件，并最终生成一个整合所有数据的主表。通过清理无效字符、增加异常处理机制、优化性能参数等方式，能够更加高效地完成数据整合工作。在具体项目中，应根据实际情况灵活调整和优化实现方案。
