Pandas批量文件处理性能优化：避免循环内concat与并发实践

import pandas as pd
import os

# 假设 root_path 和 df 已经定义
# root_path = '/path/to/your/root'
# df = pd.DataFrame({'File ID': ['folderA', 'folderB'], 'File Name': ['file001.txt', 'file002.txt']})

merged_data = pd.DataFrame()
count = 0
for index, row in df.iterrows():
    folder_name = row['File ID'].strip()
    file_name = row['File Name'].strip()
    file_path = os.path.join(root_path, folder_name, file_name)

    # 读取、转置并格式化文件数据
    file_data = pd.read_csv(file_path, names=['Case', f'{folder_name}_{file_name}'], sep='\t')
    file_data_transposed = file_data.set_index('Case').T.reset_index(drop=True)
    file_data_transposed.insert(loc=0, column='folder_file_id', value=str(folder_name+'_'+file_name))

    # 每次循环都进行 concat
    merged_data = pd.concat([merged_data, file_data_transposed], axis=0, ignore_index=True)
    count = count + 1
    print(count)

import pathlib
import pandas as pd

# 假设 root_path 和 df 已经定义
root_path = pathlib.Path('root') # 使用 pathlib 代替 os.path
df = pd.DataFrame({
    'File ID': ['folderA', 'folderB'], 
    'File Name': ['file001.txt', 'file002.txt']
})

data = {} # 用于收集所有处理后的数据
# 使用 enumerate 而非外部计数器，并直接迭代 df.iterrows()
for count, (_, row) in enumerate(df.iterrows(), 1):
    folder_name = row['File ID'].strip()
    file_name = row['File Name'].strip()

    # 使用 pathlib 构建文件路径，更简洁安全
    file_path = root_path / folder_name / file_name
    folder_file_id = f'{folder_name}_{file_name}'

    # 读取CSV文件，指定 header=None 因为文件没有表头
    # memory_map=True 可以提高大文件读取效率
    # low_memory=False 确保正确推断所有列的数据类型
    file_data = pd.read_csv(file_path, header=None, sep='\t',
                            names=['Case', folder_file_id],
                            memory_map=True, low_memory=False)

    # 设置 'Case' 列为索引，并使用 squeeze() 将单列DataFrame转换为Series
    data[folder_file_id] = file_data.set_index('Case').squeeze()
    print(count)

# 循环结束后，一次性进行 concat
merged_data = (pd.concat(data, names=['folder_file_id'])
                 .unstack('Case').reset_index())

# df
   File ID    File Name
0  folderA  file001.txt
1  folderB  file002.txt

# root/folderA/file001.txt
0   1234
1   5678
2   9012
3   3456
4   7890

# root/folderB/file002.txt
0   4567
1   8901
2   2345
3   6789

>>> merged_data
Case       folder_file_id       0       1       2       3       4
0     folderA_file001.txt  1234.0  5678.0  9012.0  3456.0  7890.0
1     folderB_file002.txt  4567.0  8901.0  2345.0  6789.0     NaN

from concurrent.futures import ThreadPoolExecutor
import pathlib
import pandas as pd

root_path = pathlib.Path('root')
df = pd.DataFrame({
    'File ID': ['folderA', 'folderB'], 
    'File Name': ['file001.txt', 'file002.txt']
})

def read_and_process_csv(args):
    """
    一个辅助函数，用于在单独的线程中读取和处理单个CSV文件。
    """
    count, row_dict = args # 解包参数，row_dict 是 df.to_dict('records') 的一行
    folder_name = row_dict['File ID'].strip()
    file_name = row_dict['File Name'].strip()
    file_path = root_path / folder_name / file_name
    folder_file_id = f'{folder_name}_{file_name}'

    file_data = pd.read_csv(file_path, header=None, sep='\t',
                            names=['Case', folder_file_id],
                            memory_map=True, low_memory=False)
    print(f"Processing {count}: {folder_file_id}")
    return folder_file_id, file_data.set_index('Case').squeeze()

# 创建一个线程池，max_workers 根据CPU核心数和I/O负载调整
with ThreadPoolExecutor(max_workers=4) as executor: # 示例使用4个工作线程
    # 将 DataFrame 转换为字典列表，以便传递给线程池
    # enumerate 用于添加计数器
    batch_args = enumerate(df[['File ID', 'File Name']].to_dict('records'), 1)

    # 使用 executor.map 并行执行 read_and_process_csv 函数
    # data 将是一个迭代器，按提交顺序返回结果
    data_iterator = executor.map(read_and_process_csv, batch_args)

    # 将迭代器转换为字典，以便 pd.concat 处理
    data_dict = dict(data_iterator)

# 循环结束后，一次性进行 concat
merged_data = (pd.concat(data_dict, names=['folder_file_id'])
                 .unstack('Case').reset_index())