本文介绍一种基于字典哈希的高效算法，用于合并多个结构相同的csv文件，以日期、时间、名称对为联合键，快速计算各数值列（number1–number7）的最大值，避免暴力嵌套循环，兼顾时间与空间效率。
本文介绍一种基于字典哈希的高效算法，用于合并多个结构相同的csv文件，以日期、时间、名称对为联合键，快速计算各数值列（number1–number7）的最大值，避免暴力嵌套循环，兼顾时间与空间效率。
在处理多源CSV数据时，常见需求是“按业务主键对齐行、跨文件聚合数值”。例如，多个传感器日志文件均含字段 date, time, name1, name2, number1,…,number7，需对每组相同 (date, time, name1, name2) 的记录，提取各 number* 列的最大值。若采用两两文件嵌套遍历（O(n×m×…）），时间复杂度将随文件数量和行数急剧上升，不可扩展。
推荐方案：单次扫描 + 字典哈希聚合
核心思想是将 (date, time, name1, name2) 作为复合键（tuple），用 Python 字典 max_values 缓存当前已见的最大值列表。逐文件、逐行读取，动态更新——既保证 O(1) 平均查找/插入，又仅需一次完整遍历所有数据，整体时间复杂度为 O(N)（N 为总行数），空间复杂度为 O(K)（K 为唯一键数量）。
以下是完整可运行示例（使用标准库 csv，兼容 Python 3.6+）：
import csv
from typing import Dict, List, Tuple, Any
def merge_csv_max(
filenames: List[str],
key_columns: int = 4,
value_columns: int = 7,
delimiter: str = ','
) -> Dict[Tuple[str, ...], List[float]]:
"""
合并多个CSV文件，按前key_columns列分组，取后续value_columns列的最大值
Args:
filenames: CSV文件路径列表
key_columns: 作为分组键的列数（默认4：date,time,name1,name2）
value_columns: 需取最大值的数值列数（默认7：number1-number7）
delimiter: CSV分隔符
Returns:
dict: 键为(key1,key2,...)，值为[value1_max, ..., value7_max]（float列表）
"""
max_values: Dict[Tuple[str, ...], List[float]] = {}
for filename in filenames:
with open(filename, 'r', newline='', encoding='utf-8') as f:
reader = csv.reader(f, delimiter=delimiter)
for row in reader:
if len(row) < key_columns + value_columns:
continue # 跳过格式异常行
# 提取键（转为tuple以支持字典索引）和数值列
key = tuple(row[:key_columns])
try:
values = [float(x) for x in row[key_columns:key_columns + value_columns]]
except ValueError:
continue # 跳过非数值内容
if key not in max_values:
max_values[key] = values.copy()
else:
# 逐列比较并更新最大值
for i in range(value_columns):
if values[i] > max_values[key][i]:
max_values[key][i] = values[i]
return max_values
# 使用示例
if __name__ == "__main__":
files = ["sensor_20240101.csv", "sensor_20240102.csv", "sensor_20240103.csv"]
result = merge_csv_max(files)
# 打印前5个结果
for i, (k, v) in enumerate(list(result.items())[:5]):
print(f"Key {k} → Max values: {v}")✅ 关键优势说明：
- 无需预加载全部文件到内存：逐行流式处理，内存占用仅取决于唯一键数量；
- 天然去重与覆盖：相同键自动合并，后出现的大值自然覆盖旧值；
- 强健性增强：添加了行长度校验、数值类型转换异常捕获，避免因脏数据中断流程；
- 灵活可配置：key_columns 和 value_columns 参数支持不同字段布局，无需修改核心逻辑。
⚠️ 注意事项：
- 确保所有CSV文件编码一致（推荐 UTF-8），并在 open() 中显式声明 encoding；
- 若字段含逗号或换行符，请改用 csv.DictReader 并指定 quoting=csv.QUOTE_MINIMAL；
- 对于超大规模数据（千万级唯一键），可考虑改用 pandas.concat(...).groupby(...).max()（需足够内存）或切换至 Dask/Polars 实现外存计算；
- 如需保留原始文件来源信息（如哪一行贡献了最大值），可在字典中额外存储元数据（如 (max_value, filename, line_no) 元组）。
最终，该方法以极简代码实现高性能聚合，是处理多文件同构数据对齐任务的典型工程实践范式——用合适的数据结构（哈希表），替代低效的算法暴力（嵌套循环）。
