用 Python 高效处理大文件

,为了进行并行处理，我们将任务划分为子单元。它增加了程序处理的作业数量，减少了整体处理时间。,例如，如果你正在处理一个大的CSV文件，你想修改一个单列。我们将把数据以数组的形式输入函数，它将根据可用的进程数量，一次并行处理多个值。这些进程是基于你的处理器内核的数量。,在这篇文章中，我们将学习如何使用multiprocessing、joblib和tqdm Python包减少大文件的处理时间。这是一个简单的教程，可以适用于任何文件、数据库、图像、视频和音频。,我们将使用来自 Kaggle 的 US Accidents (2016 - 2021) 数据集，它包括280万条记录和47个列。,https://www.kaggle.com/datasets/sobhanmoosavi/us-accidents,我们将导入multiprocessing、joblib和tqdm用于并行处理，pandas用于数据导入，re、nltk和string用于文本处理。,在我们开始之前，让我们通过加倍cpu_count()来设置n_workers。正如你所看到的，我们有8个workers。,下一步，我们将使用pandas read_csv函数读取大型CSV文件。然后打印出dataframe的形状、列的名称和处理时间。,输出：,clean_text是一个用于处理文本的简单函数。我们将使用nltk.copus获得英语停止词，并使用它来过滤掉文本行中的停止词。之后，我们将删除句子中的特殊字符和多余的空格。它将成为确定串行、并行和批处理的处理时间的基准函数。,对于串行处理，我们可以使用pandas的.apply()函数，但是如果你想看到进度条，你需要为pandas激活tqdm，然后使用.progress_apply()函数。,我们将处理280万条记录，并将结果保存回 “Description” 列中。,高端处理器串行处理280万行花了9分5秒。,有多种方法可以对文件进行并行处理，我们将了解所有这些方法。multiprocessing是一个内置的python包，通常用于并行处理大型文件。,我们将创建一个有8个workers的多处理池，并使用map函数来启动进程。为了显示进度条，我们将使用tqdm。,map函数由两部分组成。第一个部分需要函数，第二个部分需要一个参数或参数列表。,我们的处理时间几乎提高了3倍。处理时间从9分5秒下降到3分51秒。,我们现在将学习另一个Python包来执行并行处理。在本节中，我们将使用joblib的Parallel和delayed来复制map函数。,下面的过程是相当通用的，你可以根据你的需要修改你的函数和数组。我曾用它来处理成千上万的音频和视频文件，没有任何问题。,建议：使用 "try: "和 "except: "添加异常处理。,在text_parallel_clean()中添加“Description”列。,我们的函数比多进程处理Pool多花了13秒。即使如此，并行处理也比串行处理快4分59秒。,有一个更好的方法来处理大文件，就是把它们分成若干批，然后并行处理。让我们从创建一个批处理函数开始，该函数将在单一批次的值上运行clean_function。,批量处理函数,下面的函数将根据workers的数量把文件分成多个批次。在我们的例子中，我们得到8个批次。,最后，我们将使用Parallel和delayed来处理批次。,我们已经改善了处理时间。这种技术在处理复杂数据和训练深度学习模型方面非常有名。,tqdm将多处理带到了一个新的水平。它简单而强大。,process_map需要：,通过一行代码，我们得到了最好的结果：,我们需要找到一个平衡点，它可以是串行处理，并行处理，或批处理。如果你正在处理一个较小的、不太复杂的数据集，并行处理可能会适得其反。,在这个教程中，我们已经了解了各种处理大文件的Python包，它们允许我们对数据函数进行并行处理。,如果你只处理一个表格数据集，并且想提高处理性能，那么建议你尝试Dask、datatable和RAPIDS。