将每个熊猫行优化为一个不同的.txt

Question

问题

我正在处理一个有两列的df：'body'和'label'。我需要将每一行的'body'写入不同的.txt。目前，我正在通过迭代行并使用python的文件IO管理器编写它们来实现这一点，但是随着我处理的行数的增加，它变得太慢了。

下面是实际代码的情况：(的行号必须是文件名！)

for index, row in df.iterrows():
    with open(path+str(index)+".txt", "w+", encoding="utf-8") as f:
        f.write(row['body'])

我相信有一个更好的方法，我只是不知道它是什么。

进展

我已经在使用pandarallel，但没有设法在行上使用lambda函数，这样就捕获了行的索引并将其发送到aux函数。

def writer_aux(body_text, index):
    with open(path+str(index)+".txt", "w+", encoding="utf-8") as f:
        f.write(body_text)

df['body'].parallel_apply(lambda x: writer_aux(x,x.index)) # something like that

有人能帮我提出一个建议，让lambda捕获行索引，还是用另一种方式更有效地完成这一工作？谢谢

Answer 1

通过使用Python的multiprocessing，我能够显著提高性能。下面是我是如何做到的(就像阅读木星笔记本一样阅读代码)：

让我们从导入所需模块开始：

In [1]: import pandas as pd
        from timeit import default_timer as timer
        import multiprocessing as mp
        import uuid

模拟您的数据：

In [2]: num_rows = 250000
        num_hex = 100

        df = pd.DataFrame.from_dict({
            "label": [f"Some label {i}" for i in range(num_rows)],
            "body": ["".join([uuid.uuid4().hex for _ in range(num_hex)])
                     for _ in range(num_rows)]
        })

        df

Out [2]:

        label                          body
0   Some label 0    594a41b8960d4856871efef2ea39d39812ec046f175343...
1   Some label 1    9561a81d063b41298e27b079fb180e2fbaaad3768ace4b...
2   Some label 2    fabc1604c6494d56bfb7878ad5859f68a2a6f294946046...
3   Some label 3    e8aabab1ddc04b5aab1050aec7873bb3f0deec79af9e41...
4   Some label 4    08c70b7121c047838d41d6312417ea40827558b0b74342...
... ... ...
249995  Some label 249995   69ba44d138b04b0497d2457f89a54ea95ef7330fb44745...
249996  Some label 249996   4466820a1a634ff59fa3a544c7623fc27caaf1f04a264c...
249997  Some label 249997   f03d2e7087964e5f8fdb74e4799535be740f48937c4c43...
249998  Some label 249998   41dfeb037f9f44789024b766bfb864ee2bc05e9c50aa49...
249999  Some label 249999   2aa84b40276443b98a588c9677713769d2b756398d1f46...

250000 rows × 2 columns

下面是将行写入.txt文件的函数：

In [3]: def row2txt(idx, out_path="./out"):
            row = df.iloc[idx]
            file_name = f"{out_path}/row{idx}.txt"
            with open(file_name, "w+", encoding="utf-8") as f:
                f.write(row['body'])

现在，让我们尝试一下“朴素”方法，看看它做得有多好：

In [4]: # SEQUENTIAL
        start_time = timer()
        for idx in range(len(df)):
            row2txt(idx)
        print(f"[SEQUENTIAL] Elapsed time: {timer() - start_time}s")

Out [4]: [SEQUENTIAL] Elapsed time: 26.99295247200007s

现在，让我们使用Python的multiprocessing

In [5]: # PARALLEL
        start_time = timer()
        with mp.Pool() as pool:
            pool.map(row2txt, [idx for idx in range(len(df))])
        print(f"[PARALLEL] Elapsed time: {timer() - start_time}s")

Out [5]: [PARALLEL] Elapsed time: 4.735888680999778s

如您所见，并行化方法比顺序方法快5.7倍！您的计算机拥有的CPU核越多，这两种方法之间的差异就应该越大。我运行测试的机器有6个内核(当您考虑超线程时，有12个内核)。

链接 to the code (GitHub)。你也可以在Google上运行它。

Answer 2

这不会以一种有意义的方式变得更快。即使parallel_apply以并行方式运行，您也不会获得太多的好处，因为缓慢来自于文件I/O，而不是迭代。

如果要将所有行写入同一个文件(而不是为每一行编写一个新文件)，则可以通过缓冲进行一些加速，但这仍然比纯迭代慢得多。

如果parallel_apply的工作方式与df.apply相同(但并行)，则最后一行

df['body'].parallel_apply(lambda x: writer_aux(x,x.index))

应该是

df.parallel_apply(lambda x: writer_aux(x['body'], x.name), axis=1)

(在执行df['body'].apply(...)时，只有“body”传递给应用程序中的函数，而不是索引。因此，x.index将是正文文本的str.index()方法，而不是行的索引。)

你也可以：

df[['body']].apply(lambda x: writer_aux(x, x.name), axis=1)
# note the double brackets around 'body'