合并 100,000 个 csv 文件并在每次迭代中添加缺失的信息

Question

我正在编写一个代码，它将大约 100,000 个文本文件组合在一起并添加丢失的数据。每次迭代调用pandas df很慢，有更推荐的方式吗？

for root, dirs, files in os.walk(resultsDir):
    for file in files:
        if file.endswith('.txt'):
            resultsFile = os.path.relpath(os.path.join(root, file))
            df = pd.read_csv(resultsFile, delimiter='\t')
            # parsePath data
            df['path'] = resultsFile
            df['resultsFile'] = file
            pathLvls = resultsFile.split(os.sep)
            df['sdCard'] = pathLvls[-3][:5]
            df['site'] = pathLvls[-3][6:]
            df['capture'] = pathLvls[-4]
            if not os.path.isfile(resultsFile):
                print(f' resultsFile {resultsFile} is missing')

Answer 1

解决方案是使用

threading

或

asyncio

进行基于 I/O 的操作。

这是一个例子：

import pandas as pd
import threading
import queue
import time

file_list = ['file1.csv', 'file2.csv', ...] # List of n file paths

def load_file(file, queue):
    df = pd.read_csv(file)
    queue.put(df)

q = queue.Queue() 

threads = []
for f in file_list:
    t = threading.Thread(target=load_file, args=(f, q)) 
    t.start()
    threads.append(t)

df = pd.DataFrame()
for t in threads:
    t.join()
    df = df.append(q.get())
    
print(df)

基本上是这样的：

• 使用队列在线程之间共享加载的DataFrame
• 每个文件一个线程（我从未尝试过 100,000 个线程）
• 加入所有线程等待完成

您需要使用

os.walk()

来填充列表等。