Python高效处理大文件的方法详解

Jasper

发布时间：2026-10-16 | 2人浏览过

原创

本文深入解析Python处理大型数据集的性能优化方案。通过对比串行处理、多进程池（multiprocessing）、joblib并行处理及批处理技术，展示如何显著缩短280万行数据的清洗耗时。重点介绍利用tqdm实现可视化进度监控，并提供从基础到高级的并行处理代码示例与性能数据对比。

为实现并行处理，需将任务拆分为子单元。此举可增加程序并发处理的作业量，从而缩短整体处理耗时。

以处理大型CSV文件并修改单列为例，可将数据作为数组传入函数。系统依据可用进程数，同时处理多个数值，进程数量取决于处理器的内核数量。

本文将介绍如何使用multiprocessing、joblib和tqdm这三个Python库来缩短大文件的处理时间。该教程具有通用性，适用于各类文件、数据库、图像、视频及音频的处理场景。

开始

本例采用Kaggle提供的US Accidents (2016 - 2021)数据集，该数据集包含280万条记录及47个字段。

我们将导入multiprocessing、joblib和tqdm用于并行处理，pandas用于数据加载，以及re、nltk和string用于文本预处理。

# Parallel Computing
import multiprocessing as mp
from joblib import Parallel, delayed
from tqdm.notebook import tqdm
# Data Ingestion  
import pandas as pd
# Text Processing  
import re  
from nltk.corpus import stopwords
import string

在开始之前，通过将cpu_count()的结果乘以2来设置n_workers。结果显示，我们拥有8个工作进程。

n_workers = 2 * mp.cpu_count()
print(f"{n_workers} workers are available")
>>> 8 workers are available

接下来，使用pandas的read_csv函数读取大型CSV文件，随后打印DataFrame的形状、列名以及处理耗时。

%%time
file_name="../input/us-accidents/US_Accidents_Dec21_updated.csv"
df = pd.read_csv(file_name)
print(f"Shape:{df.shape}\n\nColumn Names:\n{df.columns}\n")

输出：

Shape:(2845342, 47) Column Names: Index(['ID', 'Severity', 'Start_Time', 'End_Time', 'Start_Lat', 'Start_Lng', 'End_Lat', 'End_Lng', 'Distance(mi)', 'Description', 'Number', 'Street', 'Side', 'City', 'County', 'State', 'Zipcode', 'Country', 'Timezone', 'Airport_Code', 'Weather_Timestamp', 'Temperature(F)', 'Wind_Chill(F)', 'Humidity(%)', 'Pressure(in)', 'Visibility(mi)', 'Wind_Direction', 'Wind_Speed(mph)', 'Precipitation(in)', 'Weather_Condition', 'Amenity', 'Bump', 'Crossing', 'Give_Way', 'Junction', 'No_Exit', 'Railway', 'Roundabout', 'Station', 'Stop', 'Traffic_Calming', 'Traffic_Signal', 'Turning_Loop', 'Sunrise_Sunset', 'Civil_Twilight', 'Nautical_Twilight', 'Astronomical_Twilight'], dtype='object') CPU times: user 33.9 s, sys: 3.93 s, total: 37.9 s Wall time: 46.9 s

处理文本

clean_text是一个用于文本处理的简单函数。我们利用nltk.corpus获取英语停用词，并据此过滤文本行中的停用词。随后，删除句子中的特殊字符及多余空格。该函数将作为衡量串行、并行及批处理时间的基准。

def clean_text(text):  
 # Remove stop words
 stops = stopwords.words("english")
  text = " ".join([word for word in text.split() if word  
not in stops])
 # Remove Special Characters
 text = text.translate(str.maketrans('', '', string.punctuation))
 # removing the extra spaces
 text = re.sub(' +',' ', text)
 return text

串行处理

对于串行处理，可使用pandas的.apply()函数。若需显示进度条，需为pandas启用tqdm，随后使用.progress_apply()函数。

我们将处理280万条记录，并将结果保存回“Description”列。

%%time
tqdm.pandas()
df['Description'] = df['Description'].progress_apply(clean_text)

输出

在高端处理器上，串行处理280万行数据耗时9分5秒。

100% 2845342/2845342 [09:05<00:00, 5724.25it/s] CPU times: user 8min 14s, sys: 53.6 s, total: 9min 7s Wall time: 9min 5s

多进程处理

文件并行处理有多种方法，我们将逐一探讨。multiprocessing是Python内置包，常用于大型文件的并行处理。

我们将创建一个包含8个工作进程的进程池，并使用map函数启动进程。为了显示进度条，我们将使用tqdm。

map函数由两部分组成：第一部分指定函数，第二部分指定参数或参数列表。

%%time
p = mp.Pool(n_workers)  
df['Description'] = p.map(clean_text,tqdm(df['Description']))

输出

我们的处理时间几乎提升了3倍，从9分5秒缩短至3分51秒。

100% 2845342/2845342 [02:58<00:00, 135646.12it/s] CPU times: user 5.68 s, sys: 1.56 s, total: 7.23 s Wall time: 3min 51s

并行处理

接下来学习另一个用于执行并行处理的Python库。本节将使用joblib的Parallel和delayed来复现map函数的功能。

Parallel需要两个参数：n_job = 8和backend = multiprocessing。
随后，我们将clean_text封装在delayed函数中。
创建一个循环，每次输入一个值。

以下过程具有通用性，可根据需求修改函数和数组。我曾使用该方法处理成千上万的音频和视频文件，运行无误。

建议：使用“try:”和“except:”添加异常处理。

def text_parallel_clean(array):
 result = Parallel(n_jobs=n_workers,backend="multiprocessing")(
 delayed(clean_text)
  (text)  
 for text in tqdm(array)
 )
 return result

在text_parallel_clean()中添加“Description”列。

%%time
df['Description'] = text_parallel_clean(df['Description'])

输出

我们的函数比多进程处理Pool多耗时13秒。尽管如此，并行处理仍比串行处理快4分59秒。

100% 2845342/2845342 [04:03<00:00, 10514.98it/s] CPU times: user 44.2 s, sys: 2.92 s, total: 47.1 s Wall time: 4min 4s

并行批量处理

处理大文件的一种更优方法是将其分割为若干批次，然后并行处理。首先创建一个批处理函数，该函数将在单个批次的值上运行clean_function。

批量处理函数

def proc_batch(batch):
 return [
 clean_text(text)
 for text in batch
 ]

将文件分割成批

以下函数将根据workers的数量将文件分割为多个批次。在本例中，我们得到8个批次。

def batch_file(array,n_workers):
 file_len = len(array)
 batch_size = round(file_len / n_workers)
 batches = [
 array[ix:ix+batch_size]
 for ix in tqdm(range(0, file_len, batch_size))
 ]
 return batches
batches = batch_file(df['Description'],n_workers)
>>> 100% 8/8 [00:00<00:00, 280.01it/s]

运行并行批处理

最后，我们将使用Parallel和delayed来处理这些批次。

%%time
batch_output = Parallel(n_jobs=n_workers,backend="multiprocessing")(
 delayed(proc_batch)
  (batch)  
 for batch in tqdm(batches)
 )
df['Description'] = [j for i in batch_output for j in i]

输出

我们优化了处理时间。该技术常用于处理复杂数据及训练深度学习模型。

100% 8/8 [00:00<00:00, 2.19it/s] CPU times: user 3.39 s, sys: 1.42 s, total: 4.81 s Wall time: 3min 56s

tqdm 并发

tqdm将多处理提升到了新的高度，它简单而强大。

process_map需要：

函数名称
Dataframe 列名
max_workers
chucksize与批次大小类似。我们将根据workers的数量计算批处理大小，或可根据个人喜好添加该数值。

%%time
from tqdm.contrib.concurrent import process_map
batch = round(len(df)/n_workers)
df['Description'] = process_map(clean_text,df['Description'], max_workers=n_workers, chunksize=batch)

输出

通过一行代码，我们得到了最佳结果：