本教程详细介绍了如何使用python的`pathlib`模块高效遍历嵌套子文件夹，结合`pandas`库读取并合并散布在不同位置的多个csv文件，最终生成一个统一的、包含所有数据的csv主文件。文章将提供清晰的步骤、优化的代码示例及注意事项，帮助用户轻松实现复杂文件结构下的数据整合。
在数据分析和处理的场景中，我们经常会遇到数据分散存储在多个文件甚至多层嵌套的文件夹中的情况。例如，日志文件按日期或事件类型存储在不同的子目录中，或者实验数据按批次或条件存储。手动查找并合并这些文件不仅耗时，而且容易出错。本文将指导您如何利用Python的pathlib和pandas库，自动化地从复杂的文件结构中提取并合并CSV数据，生成一个统一的、可供进一步分析的主数据集。
1. 环境准备
在开始之前，请确保您的Python环境中已安装pandas库。如果尚未安装，可以通过以下命令进行安装：
pip install pandas
2. 理解文件结构与目标
假设我们有一个名为 Sessions 的父目录，其中包含多个子文件夹，如 day1, day2, day3 等。每个 dayX 文件夹下又有一个 weather 子文件夹，其中包含一个或多个 weather*.csv 文件。我们的目标是遍历 Sessions 目录下的所有 weather*.csv 文件，将它们的内容合并到一个名为 weather_All.csv 的主文件中，并将其保存在 Sessions/Weather/ 目录下。
示例文件结构如下：
Sessions/ ├── day1/ │ └── weather/ │ └── weather1.csv ├── day2/ │ └── weather/ │ └── weather2.csv ├── day3/ │ └── weather/ │ └── weather3.csv └── ...
目标输出文件：
Sessions/
└── Weather/
└── weather_All.csv3. 核心解决方案：pathlib与pandas
Python的pathlib模块提供了面向对象的文件系统路径操作，使得路径处理更加直观和强大。rglob()方法尤其适用于递归查找指定模式的文件。结合pandas库的数据读取和合并能力，我们可以高效地完成任务。
3.1 查找所有目标CSV文件
pathlib.Path.rglob()方法允许我们递归地搜索指定目录及其所有子目录中符合特定模式的文件。在这里，我们将使用它来查找所有以.csv结尾的文件。
from pathlib import Path
import pandas as pd
# 定义父目录和目标输出文件路径
parent_directory = 'Sessions'
output_dir = Path(parent_directory) / 'Weather'
output_file = output_dir / 'weather_All.csv'
# 确保输出目录存在，如果不存在则创建
output_dir.mkdir(parents=True, exist_ok=True)
# 用于存储所有DataFrame的列表
all_dfs = []
# 使用rglob递归查找所有.csv文件
print(f"开始在 '{parent_directory}' 中查找CSV文件...")
for file_path in Path(parent_directory).rglob('*.csv'):
# 过滤掉目标输出文件本身，防止将其读入并合并
if file_path == output_file:
continue
try:
# 读取CSV文件
df = pd.read_csv(file_path)
all_dfs.append(df)
print(f"已读取文件: {file_path}")
except Exception as e:
print(f"读取文件 '{file_path}' 时发生错误: {e}")
# 根据需要决定是跳过错误文件还是中断程序
print("所有CSV文件读取完毕。")3.2 合并DataFrame并保存
收集到所有单个CSV文件的DataFrame后，我们可以使用pd.concat()函数一次性将它们合并成一个大的DataFrame。这种方法比在循环中反复使用concat更高效，因为它避免了多次创建新的DataFrame对象。
# 检查是否有DataFrame需要合并
if all_dfs:
# 使用pd.concat合并所有DataFrame
# ignore_index=True 会重置合并后的DataFrame的索引
combined_df = pd.concat(all_dfs, ignore_index=True)
# 将合并后的DataFrame保存为新的CSV文件
# index=False 避免将DataFrame的索引写入CSV文件
# encoding='utf-8-sig' 确保在Windows等系统上正确处理中文字符和BOM
combined_df.to_csv(output_file, index=False, encoding='utf-8-sig')
print(f"\n成功合并 {len(all_dfs)} 个CSV文件，并保存至 '{output_file}'")
else:
print(f"\n在 '{parent_directory}' 中未找到任何CSV文件进行合并。")
4. 完整代码示例
将上述步骤整合，得到完整的解决方案代码：
from pathlib import Path
import pandas as pd
def combine_nested_csvs(parent_directory: str, output_filename: str = 'weather_All.csv'):
"""
从指定父目录及其所有子目录中查找并合并所有CSV文件。
Args:
parent_directory (str): 包含CSV文件的根目录。
output_filename (str): 合并后输出的CSV文件名。
该文件将保存在 parent_directory/Weather/ 目录下。
"""
# 定义父目录和目标输出文件路径
base_path = Path(parent_directory)
output_dir = base_path / 'Weather'
output_file = output_dir / output_filename
# 确保输出目录存在，如果不存在则创建
output_dir.mkdir(parents=True, exist_ok=True)
# 用于存储所有DataFrame的列表
all_dfs = []
print(f"--- 开始在 '{base_path}' 中查找并合并CSV文件 ---")
# 使用rglob递归查找所有.csv文件
for file_path in base_path.rglob('*.csv'):
# 过滤掉目标输出文件本身，防止将其读入并合并
if file_path == output_file:
print(f"跳过目标输出文件: {file_path}")
continue
try:
# 读取CSV文件
df = pd.read_csv(file_path)
all_dfs.append(df)
print(f"已读取文件: {file_path}")
except pd.errors.EmptyDataError:
print(f"警告: 文件 '{file_path}' 为空，已跳过。")
except Exception as e:
print(f"错误: 读取文件 '{file_path}' 时发生异常: {e}")
# 根据需要决定是跳过错误文件还是中断程序
# 检查是否有DataFrame需要合并
if all_dfs:
print(f"\n--- 正在合并 {len(all_dfs)} 个DataFrame ---")
# 使用pd.concat合并所有DataFrame
combined_df = pd.concat(all_dfs, ignore_index=True)
# 将合并后的DataFrame保存为新的CSV文件
combined_df.to_csv(output_file, index=False, encoding='utf-8-sig')
print(f"\n--- 成功合并所有CSV文件，并保存至 '{output_file}' ---")
print(f"合并后的DataFrame包含 {len(combined_df)} 行数据。")
else:
print(f"\n--- 在 '{base_path}' 中未找到任何CSV文件进行合并。---")
# 调用函数执行合并操作
if __name__ == "__main__":
# 假设您的父目录名为 'Sessions'
combine_nested_csvs('Sessions', 'weather_All.csv')
5. 注意事项与最佳实践
- 列名一致性： 确保所有待合并的CSV文件具有相同或兼容的列名和数据类型。pd.concat会根据列名进行对齐，如果列名不一致，可能会引入额外的列并用NaN填充。
- 文件编码： 在读取和写入CSV文件时，指定正确的编码（如encoding='utf-8-sig'）可以避免乱码问题，特别是处理包含非英文字符或在不同操作系统（如Windows）之间传输的文件时。
- 错误处理： 示例代码中包含了try-except块来捕获读取CSV文件时可能发生的错误（如文件损坏、格式不正确或为空）。您可以根据实际需求增强错误处理逻辑，例如记录错误日志或跳过特定文件。
- 内存效率： 对于数量巨大或单个文件非常大的情况，将所有DataFrame一次性加载到内存中可能会导致内存溢出。在这种情况下，可以考虑使用dask等库进行并行处理，或者逐块读取和写入数据。
- 目标文件过滤： 务必在遍历文件时跳过最终的合并输出文件，否则程序可能会尝试读取自身，导致无限循环或错误。
- ignore_index=True： 在pd.concat中使用ignore_index=True是一个好习惯，它会为合并后的DataFrame生成一个新的连续索引，避免原始文件索引的重复或冲突。
- 实际复制文件： 本教程的目的是合并数据，而不是复制文件。如果您确实需要将所有单独的CSV文件复制到一个目标目录，可以在读取每个文件后，使用shutil.copy()函数进行复制操作。
总结
通过结合pathlib的强大文件系统导航能力和pandas高效的数据处理功能，我们可以轻松应对从复杂目录结构中聚合数据的挑战。本文提供的解决方案不仅自动化了这一过程，而且考虑了效率和健壮性，是处理类似数据整合任务的专业级实践。
