读取特定文件夹中的所有 csv 文件，合并它们，并找到最大值 w.r.t。行间隔

Question

我有 120 个 .csv 文件。包括IndexNo、日期、EArray、温度等。

这里的索引列从 1 到 8760 不等。 我想从文件夹中读取所有 csv 文件并将它们合并到单个文件中。 合并这些文件后，我将拥有所有 IndexNo 120 次（即 IndexNo 1 将有 120 行）。

在此之后，我必须为每个 IndexNo（即 IndexNo 1 到 8760）找到 EArray 的最大值并打印该最大 EArray 值行。

import pandas , OS, 
glob path = r'C:\Data_Input' # use your path 
all_files = glob.glob(path + "/*.csv") 
# print(all_files) 
li = [] 
for filename in all_files: 
     df = pd.read_csv(filename, skiprows=10, names=None, engine='python',header=0, encoding='unicode_escape') 
     df = df.assign(File_name=os.path.basename(filename).split('.')[0]) 
     li.append(df) 
     frame = pd.concat(li, axis=0, ignore_index=True, sort=False)


frame = frame.dropna() 
df = frame.assign(max_EArray=frame.groupby('IndexNo')['EArray'].transform('max')) df_filtered = df[df['EArray'] == df['max_EArray']] 
output = df_filtered.loc[df_filtered.ne(0).all(axis=1)]('max_EArray', axis=1) print(output.shape) 
output.to_csv('temp.csv') 

Answer 1

使用dask（而不是纯粹的Pandas）可以轻松完成您的任务。

其中一个优点是“开箱即用”您有可能获得 已从中读取特定行的源文件的名称。

我的解决方法如下：

1. 安装dask（如果您尚未安装）。

2. 导入dask.dataframe：

   import dask.dataframe as dd
   

3. 定义一个函数来重新格式化 DataFrame（单独调用 从特定 .csv 文件中读取的每个“部分”DataFrame）：

   def reformat(df):
       df.path = df.path.str.extract(r'/(\w+)\.\w+')
       return df[['IndexNo', 'EArray', 'path']]
   

   在这里您可以使用“普通”Pandas 代码。它也改变了路径， 剥离目录路径，只留下文件名（不带扩展名）。

4. 定义一个函数以从每个组中获取“最大”行（分组后 由索引号）：

   def getMax(grp):
       wrk = grp.reset_index(drop=True)
       ind = wrk.EArray.idxmax()
       return wrk.loc[ind, ['EArray', 'path']]
   

5. 运行实际处理：

   ddf = dd.read_csv('EArray/*.csv', include_path_column=True)
   ddf = ddf.map_partitions(reformat)
   ddf = ddf.groupby('IndexNo').apply(getMax, meta={'EArray': 'i4', 'path': 'O'})
   df = ddf.compute().sort_index().reset_index()
   

说明：

• 'EArray/*.csv' - 一堆源文件的规范。 我将所有源文件放在一个专用的子文件夹 (EArray) 中。
• include_path_column=True - 将 path 列添加到 DataFrame，包含 已读取每一行的文件的完整路径。
• map_partitions(...) - 分别调用 reformat 函数 “部分”数据帧。
• groupby(...) 和 apply(...) - 通常，就像在 Pandas 中一样。
• meta - dask 中需要附加参数（名称规范 以及输出 DataFrame 中的列类型）。
• compute() - 运行由前面的指令准备的处理树。 现在结果是“正常”Pandas DataFrame。
• sort_index() 和 reset_index() - Pandas 对 compute() 结果的操作。

为了测试，我准备了 3 个 .csv 文件，每个文件有 10 行：

T1.csv：

   IndexNo        date  EArray
0     1001  2019-01-01      20
1     1002  2019-01-02      20
2     1003  2019-01-03      20
3     1004  2019-01-04      20
4     1005  2019-01-05      20
5     1006  2019-01-06      20
6     1007  2019-01-07      20
7     1008  2019-01-08      20
8     1009  2019-01-09      20
9     1010  2019-01-10      20

T2.csv：

   IndexNo        date  EArray
0     1001  2019-01-11      22
1     1002  2019-01-12      23
2     1003  2019-01-13      24
3     1004  2019-01-14      25
4     1005  2019-01-15      26
5     1006  2019-01-16      27
6     1007  2019-01-17      28
7     1008  2019-01-18      29
8     1009  2019-01-19      30
9     1010  2019-01-20      31

T3.csv：

   IndexNo        date  EArray
0     1001  2019-01-21      35
1     1002  2019-01-22      34
2     1003  2019-01-23      33
3     1004  2019-01-24      32
4     1005  2019-01-25      31
5     1006  2019-01-26      30
6     1007  2019-01-27      29
7     1008  2019-01-28      28
8     1009  2019-01-29      28
9     1010  2019-01-30      26

我的程序的结果是：

   IndexNo  EArray path
0     1001      35   T3
1     1002      34   T3
2     1003      33   T3
3     1004      32   T3
4     1005      31   T3
5     1006      30   T3
6     1007      29   T3
7     1008      29   T2
8     1009      30   T2
9     1010      31   T2

例如对于 IndexNo == 1001，EArray 的值为： 20、22 和 35 foreach 输入文件。

IndexNo == 1001 的结果包含：

• EArray == 35 - 上面 3 的最大值，
• T3 - 包含“max”行的源文件。

我知道你必须学习 dask，但在我看来 值得为此付出一些努力。

请注意，我的代码非常清晰简洁。 函数只有 7 行，主程序只有 4 行。