要高效地将文件夹内所有文件（特别是Excel、CSV等数据文件）的数据读取并导入数据库或进行合并处理，可以分解为几个核心步骤：遍历文件、读取数据、数据清洗/转换、以及最终入库或合并。下面将结合具体代码示例进行详细说明。 ### 1. 遍历文件夹，获取所有目标文件路径 首先，需要遍历指定文件夹，筛选出需要处理的文件（如 `.xlsx`, `.csv` 等）。可以使用 `os` 或 `pathlib` 模块。 ```python import os import pandas as pd def get_file_paths(folder_path, file_extensions=('.xlsx', '.xls', '.csv')): """ 获取文件夹下所有指定扩展名的文件路径。 :param folder_path: 文件夹路径 :param file_extensions: 需要筛选的文件扩展名元组 :return: 文件路径列表 """ file_paths = [] for root, dirs, files in os.walk(folder_path): for file in files: if file.lower().endswith(file_extensions): file_paths.append(os.path.join(root, file)) return file_paths # 示例：获取当前目录下所有Excel和CSV文件 folder = './data' all_files = get_file_paths(folder) print(f"找到 {len(all_files)} 个文件: {all_files}") ``` ### 2. 读取文件数据 根据文件扩展名，使用 `pandas` 库的 `read_excel` 或 `read_csv` 函数读取数据。`pandas` 能自动处理大多数常见的数据格式，是数据读取的首选工具 [ref_4][ref_5]。 ```python def read_data_from_files(file_paths): """ 读取文件路径列表中的所有文件，返回一个包含DataFrame的列表。 :param file_paths: 文件路径列表 :return: DataFrame列表 """ data_frames = [] for file_path in file_paths: try: if file_path.lower().endswith(('.xlsx', '.xls')): # 读取Excel文件 df = pd.read_excel(file_path, engine='openpyxl') # 对于.xlsx文件 elif file_path.lower().endswith('.csv'): # 读取CSV文件 df = pd.read_csv(file_path, encoding='utf-8') # 根据实际编码调整 else: print(f"跳过不支持的文件格式: {file_path}") continue data_frames.append(df) print(f"成功读取: {file_path}, 数据形状: {df.shape}") except Exception as e: print(f"读取文件 {file_path} 时出错: {e}") return data_frames # 示例：读取所有文件 dfs = read_data_from_files(all_files) ``` ### 3. 数据合并 如果需要将所有文件的数据合并成一个大的 `DataFrame`，可以使用 `pandas.concat` 或循环 `append`（注意：`append` 方法在较新版本的pandas中已被弃用，推荐使用 `concat`）[ref_4]。 ```python def merge_dataframes(data_frames): """ 合并多个DataFrame。 :param data_frames: DataFrame列表 :return: 合并后的单个DataFrame """ if not data_frames: print("没有数据可以合并。") return pd.DataFrame() # 使用concat进行合并，ignore_index=True会重置索引 merged_df = pd.concat(data_frames, ignore_index=True) print(f"合并完成，总数据量: {merged_df.shape}") return merged_df # 示例：合并所有读取的DataFrame combined_data = merge_dataframes(dfs) ``` ### 4. 数据清洗与预处理 在合并或入库前，通常需要进行数据清洗，例如处理缺失值、统一列名、转换数据类型等。 ```python def clean_data(df): """ 进行基本的数据清洗。 :param df: 输入的DataFrame :return: 清洗后的DataFrame """ # 1. 统一列名（去除前后空格，转为小写） df.columns = df.columns.str.strip().str.lower() # 2. 处理缺失值（这里示例为用0填充数值列，用'未知'填充字符列） # 首先区分数值列和非数值列 numeric_cols = df.select_dtypes(include=['number']).columns non_numeric_cols = df.select_dtypes(exclude=['number']).columns df[numeric_cols] = df[numeric_cols].fillna(0) df[non_numeric_cols] = df[non_numeric_cols].fillna('未知') # 3. 转换日期列（假设有一列名为'date'） if 'date' in df.columns: df['date'] = pd.to_datetime(df['date'], errors='coerce') # 错误日期转为NaT # 处理转换后可能产生的NaT（缺失日期） df['date'].fillna(pd.Timestamp('1900-01-01'), inplace=True) print("数据清洗完成。") return df # 示例：清洗合并后的数据 cleaned_data = clean_data(combined_data) ``` ### 5. 导入数据库 将处理好的数据导入数据库（以SQL Server为例，使用 `pymssql` 或 `sqlalchemy`）。这里展示使用 `sqlalchemy` 创建引擎并批量插入数据的方法，它更通用且能更好地处理数据类型映射 [ref_1][ref_2][ref_3]。 ```python from sqlalchemy import create_engine import urllib def import_to_sql_server(df, table_name, server, database, username, password): """ 将DataFrame数据导入SQL Server数据库。 :param df: 要导入的DataFrame :param table_name: 目标表名 :param server: 服务器地址 :param database: 数据库名 :param username: 用户名 :param password: 密码 """ # 构建连接字符串 (SQL Server) # 使用SQLAlchemy的连接字符串格式 params = urllib.parse.quote_plus( f"DRIVER={{ODBC Driver 17 for SQL Server}};" f"SERVER={server};" f"DATABASE={database};" f"UID={username};" f"PWD={password}" ) connection_string = f"mssql+pyodbc:///?odbc_connect={params}" try: # 创建数据库引擎 engine = create_engine(connection_string) # 将DataFrame写入数据库表，如果表存在则替换（也可以使用'append'模式追加） df.to_sql(table_name, engine, if_exists='replace', index=False) print(f"数据成功导入到表 {table_name} 中。") except Exception as e: print(f"导入数据库时出错: {e}") # 示例：将清洗后的数据导入SQL Server # 请替换为实际的数据库连接信息 server = "your_server" database = "your_database" username = "your_username" password = "your_password" table_name = "your_table" import_to_sql_server(cleaned_data, table_name, server, database, username, password) ``` ### 6. 完整流程整合 将以上步骤整合成一个完整的脚本，实现从文件夹读取数据到入库的自动化流程。 ```python import os import pandas as pd from sqlalchemy import create_engine import urllib def main(folder_path, db_config, table_name): """ 主函数：遍历文件夹、读取数据、合并、清洗、入库。 :param folder_path: 数据文件夹路径 :param db_config: 数据库配置字典，包含server, database, username, password :param table_name: 目标表名 """ print("开始处理...") # 步骤1: 获取所有数据文件路径 file_paths = get_file_paths(folder_path) if not file_paths: print("未找到任何数据文件。") return # 步骤2: 读取所有文件数据 dfs = read_data_from_files(file_paths) if not dfs: print("没有成功读取任何数据。") return # 步骤3: 合并数据 combined_data = merge_dataframes(dfs) # 步骤4: 数据清洗 cleaned_data = clean_data(combined_data) # 步骤5: 导入数据库 import_to_sql_server(cleaned_data, table_name, **db_config) print("全部流程执行完毕。") # 配置参数 data_folder = "./data" # 你的数据文件夹路径 database_config = { 'server': 'localhost', 'database': 'TestDB', 'username': 'sa', 'password': 'your_password' } target_table = 'ImportedData' # 执行主流程 if __name__ == "__main__": main(data_folder, database_config, target_table) ``` ### 关键要点与注意事项 | 步骤 | 核心工具/方法 | 注意事项与常见问题 | | :--- | :--- | :--- | | **文件遍历** | `os.walk`, `pathlib` | 确保路径正确；注意文件权限；可扩展筛选逻辑。 | | **数据读取** | `pandas.read_excel`, `pandas.read_csv` | Excel文件需安装 `openpyxl` 或 `xlrd` 引擎；CSV文件需注意编码（如 `utf-8`, `gbk`）。 | | **数据合并** | `pandas.concat` | 确保各文件数据结构（列名、类型）一致；大数据量时注意内存使用。 | | **数据清洗** | `pandas` 数据操作（`fillna`, `astype`, `to_datetime`等） | 根据业务逻辑处理缺失值、异常值；统一数据格式。 | | **数据库导入** | `sqlalchemy.create_engine`, `DataFrame.to_sql` | 确保数据库驱动已安装；长连接需管理连接池；大批量插入可考虑分块（`chunksize`参数）。 | 此方案提供了一个从文件夹读取多种格式数据、进行初步清洗合并、并导入SQL Server数据库的完整、模块化的Python实现。通过函数封装，各步骤清晰且易于维护与扩展，可适应大多数中小规模的数据自动化导入场景 [ref_1][ref_2][ref_3][ref_4][ref_5]。对于超大规模数据（如千万条），需考虑分块读取与写入，并优化数据库连接与提交策略以避免内存不足和提升效率 [ref_6]。