问题背景
在数据处理过程中，我们经常会遇到需要处理大量结构相同的 CSV 文件的情况。例如，你可能有 1000 多个 CSV 文件，每个文件都包含需要分析的数据，并且文件名与主数据表中的键相关联。我们需要将这些文件读取到数据库中，并创建一个主表，以便进行更复杂的查询和分析。
解决方案
以下步骤将详细介绍如何使用 Python 的 pandas 库和 sqlite3 模块来解决这个问题：
-
导入必要的库：
import pandas as pd import sqlite3 import os import re
-
设置 CSV 文件目录和数据库连接：
立即学习“Python免费学习笔记（深入）”；
# 设置 CSV 文件所在的目录 csv_directory = r"path" # 将 "path" 替换为你的 CSV 文件目录 # 连接到 SQLite 数据库 (如果不存在则创建) conn = sqlite3.connect("db.db") cursor = conn.cursor() -
获取 CSV 文件列表：
# 获取目录中所有 CSV 文件的列表 csv_files = [file for file in os.listdir(csv_directory) if file.endswith(".csv")] -
循环遍历 CSV 文件并创建表：
for csv_file in csv_files: table_name = os.path.splitext(csv_file)[0] # 从文件名中提取表名 (不包含扩展名) csv_path = os.path.join(csv_directory, csv_file) # 读取 CSV 文件到 pandas DataFrame df = pd.read_csv(csv_path) # 添加包含文件名的列 df['filename'] = os.path.basename(csv_file) # 将 DataFrame 写入 SQLite 数据库 df.to_sql(table_name, conn, index=False, if_exists='replace') # 使用 replace 避免重复创建表 -
创建主表并合并数据：
# 定义主表名称 master_bus = "master_table" # 清理表名，移除无效字符 master_bus = re.sub(r'\W ', '', master_bus) # 获取所有表名 tables = cursor.execute("SELECT name FROM sqlite_master WHERE type='table';").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("No tables found to create master table.") # 提交更改并关闭连接 conn.commit() conn.close()
代码解释
- pandas.read_csv(csv_path): 使用 pandas 库读取 CSV 文件到 DataFrame 对象。
- df['filename'] = os.path.basename(csv_file): 创建一个名为 filename 的新列，并将文件名添加到该列中。
- df.to_sql(table_name, conn, index=False, if_exists='replace'): 将 DataFrame 写入 SQLite 数据库。index=False 阻止写入 DataFrame 的索引。if_exists='replace' 确保如果表已存在，则先删除该表，然后再创建。
- re.sub(r'\W ', '', master_bus): 使用正则表达式清理表名，移除所有非字母数字字符。这可以防止 SQLite 语法错误。
- *`cursor.execute(f"CREATE TABLE {master_bus} AS SELECT FROM {tables[0]} WHERE 0;")**: 创建一个与第一个表具有相同结构的主表。WHERE 0` 子句确保只创建表结构，而不插入任何数据。
- *`cursor.execute(f"INSERT INTO {master_bus} SELECT FROM {table};")`**: 将数据从每个单独的表插入到主表中。
注意事项
- 表名有效性： 确保表名不包含任何特殊字符或空格。使用 re.sub() 函数可以有效地清理表名。
- 数据类型一致性： 确保所有 CSV 文件中的数据类型一致。否则，在合并数据时可能会出现问题。
- 内存占用： 如果 CSV 文件非常大，一次性加载所有文件可能会导致内存问题。可以考虑使用 chunksize 参数分块读取 CSV 文件。
- 错误处理： 在生产环境中，应该添加适当的错误处理机制，例如 try-except 块，以处理文件读取、数据库连接等过程中可能出现的异常。
- 性能优化： 对于非常大的数据集，可以考虑使用批量插入来提高性能。
总结
本文提供了一个使用 Python 和 SQLite 处理大量 CSV 文件的完整解决方案。通过读取 CSV 文件、添加文件名列、创建单独的表，并将它们合并到一个主表中，可以方便地进行数据分析和查询。同时，注意表名有效性、数据类型一致性、内存占用和错误处理等问题，可以确保代码的稳定性和效率。通过掌握这些技巧，你可以更有效地处理大型数据集，并从中提取有价值的信息。
