Python中幾種高效讀取大文件的完整指南
更新時間：2025年06月30日 11:14:46 作者：北辰alk
處理大型文件時,我們需要采用特殊的技術(shù)來避免內(nèi)存溢出,本文主要介紹了Python中幾種高效讀取大文件的完整指南,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
在處理大型文件時（如內(nèi)存只有4G卻要讀取8G的文件），我們需要采用特殊的技術(shù)來避免內(nèi)存溢出。以下是幾種高效讀取大文件的方法。
一、基本方法：逐行讀取
1. 使用文件對象的迭代器
最簡單的方法是直接迭代文件對象，Python會自動使用緩沖IO以高效的方式處理：
with open('large_file.txt', 'r', encoding='utf-8') as f:
for line in f: # 逐行讀取，內(nèi)存友好
process_line(line) # 處理每一行
2. 明確使用 readline()
with open('large_file.txt', 'r') as f:
while True:
line = f.readline()
if not line: # 到達文件末尾
break
process_line(line)
二、分塊讀取方法
對于非文本文件或需要按塊處理的情況：
1. 指定緩沖區(qū)大小
BUFFER_SIZE = 1024 * 1024 # 1MB的緩沖區(qū)
with open('large_file.bin', 'rb') as f:
while True:
chunk = f.read(BUFFER_SIZE)
if not chunk: # 文件結(jié)束
break
process_chunk(chunk)
2. 使用 iter 和 partial
更Pythonic的分塊讀取方式：
from functools import partial
chunk_size = 1024 * 1024 # 1MB
with open('large_file.bin', 'rb') as f:
for chunk in iter(partial(f.read, chunk_size), b''):
process_chunk(chunk)
三、內(nèi)存映射文件 (mmap)
對于需要隨機訪問的大型文件：
import mmap
with open('large_file.bin', 'r+b') as f:
# 創(chuàng)建內(nèi)存映射
mm = mmap.mmap(f.fileno(), 0)
# 像操作字符串一樣操作文件
print(mm[:100]) # 讀取前100字節(jié)
# 可以搜索內(nèi)容
index = mm.find(b'some_pattern')
if index != -1:
print(f"Found at position {index}")
mm.close() # 記得關(guān)閉映射
四、使用生成器處理
將文件處理邏輯封裝為生成器：
def read_large_file(file_path, chunk_size=1024*1024):
"""生成器函數(shù)，逐塊讀取大文件"""
with open(file_path, 'rb') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
# 使用生成器
for chunk in read_large_file('huge_file.bin'):
process_chunk(chunk)
五、處理壓縮文件
對于大型壓縮文件，可以使用流式解壓：
1. gzip 文件
import gzip
import shutil
with gzip.open('large_file.gz', 'rb') as f_in:
with open('large_file_extracted', 'wb') as f_out:
shutil.copyfileobj(f_in, f_out) # 流式復制
2. zip 文件
import zipfile
with zipfile.ZipFile('large_file.zip', 'r') as z:
with z.open('file_inside.zip') as f:
for line in f:
process_line(line)
六、多線程/多進程處理
對于需要并行處理的情況：
1. 多線程處理不同塊
from concurrent.futures import ThreadPoolExecutor
import os
def process_chunk(start, end, file_path):
"""處理文件的指定部分"""
with open(file_path, 'rb') as f:
f.seek(start)
chunk = f.read(end - start)
# 處理chunk...
def parallel_file_processing(file_path, num_threads=4):
file_size = os.path.getsize(file_path)
chunk_size = file_size // num_threads
with ThreadPoolExecutor(max_workers=num_threads) as executor:
futures = []
for i in range(num_threads):
start = i * chunk_size
end = start + chunk_size if i != num_threads - 1 else file_size
futures.append(executor.submit(process_chunk, start, end, file_path))
# 等待所有任務完成
for future in concurrent.futures.as_completed(futures):
future.result()
七、使用第三方庫
1. Dask - 用于超大型數(shù)據(jù)集
import dask.dataframe as dd
# 創(chuàng)建延遲計算的DataFrame
df = dd.read_csv('very_large_file.csv', blocksize=25e6) # 25MB每塊
# 執(zhí)行操作（惰性計算）
result = df.groupby('column').mean().compute() # 實際計算
2. PyTables - 處理HDF5格式
import tables
# 打開HDF5文件
h5file = tables.open_file('large_data.h5', mode='r')
# 訪問數(shù)據(jù)集
table = h5file.root.data.table
for row in table.iterrows(): # 迭代訪問
process_row(row)
h5file.close()
八、數(shù)據(jù)庫替代方案
對于需要頻繁查詢的大型數(shù)據(jù)，考慮使用數(shù)據(jù)庫：
1. SQLite
import sqlite3
# 將數(shù)據(jù)導入SQLite
conn = sqlite3.connect(':memory:') # 或磁盤數(shù)據(jù)庫
cursor = conn.cursor()
cursor.execute('CREATE TABLE data (col1, col2, col3)')
# 批量插入數(shù)據(jù)
with open('large_file.csv') as f:
# 使用生成器避免內(nèi)存問題
data_gen = (line.strip().split(',') for line in f)
cursor.executemany('INSERT INTO data VALUES (?, ?, ?)', data_gen)
conn.commit()
九、性能優(yōu)化技巧
緩沖區(qū)大小選擇：
- 通常8KB到1MB之間效果最好
- 可通過實驗找到最佳大小
二進制模式 vs 文本模式：
- 二進制模式(
'rb')通常更快 - 文本模式(
'r')需要處理編碼，但更方便
- 二進制模式(
操作系統(tǒng)緩存：
- 現(xiàn)代OS會自動緩存頻繁訪問的文件部分
- 多次讀取同一大文件時，第二次會快很多
避免不必要的處理：
- 盡早過濾掉不需要的數(shù)據(jù)
- 使用生成器保持內(nèi)存效率
十、完整示例：處理超大CSV文件
import csv
from collections import namedtuple
from itertools import islice
def process_large_csv(file_path, batch_size=10000):
"""分批處理大型CSV文件"""
# 定義行結(jié)構(gòu)
CSVRow = namedtuple('CSVRow', ['id', 'name', 'value'])
with open(file_path, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
headers = next(reader) # 跳過標題行
while True:
# 讀取一批行
batch = list(islice(reader, batch_size))
if not batch:
break # 文件結(jié)束
# 處理批次
rows = [CSVRow(*row) for row in batch]
process_batch(rows)
# 可選：顯示進度
print(f"Processed {len(batch)} rows")
def process_batch(rows):
"""處理一批數(shù)據(jù)"""
# 這里添加實際處理邏輯
pass
# 使用
process_large_csv('huge_dataset.csv')
十一、總結(jié)
處理大文件的關(guān)鍵原則：
- 不要一次性加載到內(nèi)存：始終使用迭代或分塊方式
- 選擇合適的數(shù)據(jù)結(jié)構(gòu)：根據(jù)需求選擇逐行、分塊或內(nèi)存映射
- 考慮并行處理：對于CPU密集型處理
- 利用生成器：保持內(nèi)存效率
- 考慮專業(yè)工具：如Dask、PyTables等
通過以上技術(shù)，即使內(nèi)存有限，也能高效處理遠大于內(nèi)存的文件。記住，正確的I/O策略可以顯著影響程序性能，特別是對于大型數(shù)據(jù)集。
到此這篇關(guān)于Python中幾種高效讀取大文件的完整指南的文章就介紹到這了,更多相關(guān)Python 讀取大文件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Spectral?clustering譜聚類算法的實現(xiàn)代碼
譜聚類是從圖論中演化出來的算法，它的主要思想是把所有的數(shù)據(jù)看做空間中的點，這些點之間可以用邊連接起來，今天通過本文給大家介紹Spectral?clustering譜聚類算法的實現(xiàn)，感興趣的朋友一起看看吧2022-04-04
Python實現(xiàn)輕松處理兩行表頭Excel并可視化分析(附完整代碼)
在日常的數(shù)據(jù)處理中,我們經(jīng)常會遇到多級表頭的 Excel 文件,本文通過一個實際案例,帶你完整了解讀取 Excel ,整理多級表頭 ,提取庫存,銷量的數(shù)據(jù) ,繪圖可視化的完整步驟吧2025-11-11
這篇文章主要介紹了Python中處理字符串之islower()方法的使用,是Python入門的基礎(chǔ)知識,需要的朋友可以參考下2015-05-05
python 遍歷pd.Series的index和value
今天小編就為大家分享一篇python 遍歷pd.Series的index和value，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
用Python爬取LOL所有的英雄信息以及英雄皮膚的示例代碼
這篇文章主要介紹了用Python爬取LOL所有的英雄信息以及英雄皮膚的示例代碼，主要分為兩部分，獲取網(wǎng)頁上數(shù)據(jù)和圖片保存到本地等，感興趣的可以了解一下2020-07-07
下面小編就為大家?guī)硪黄狿ython實現(xiàn)好友全頭像的拼接實例(推薦)。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-06-06
在Python中，字符串是不可變類型，即無法直接修改字符串的某一位字符。這篇文章主要介紹了Python中修改字符串的四種方法,需要的朋友可以參考下2018-11-11
這篇文章主要介紹了教你怎樣用python編寫腳本實現(xiàn)自動簽到，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-08-08
model.eval()將模型切換至評估模式,關(guān)閉Dropout并固定BatchNorm行為,確保推理結(jié)果穩(wěn)定準確,測試、驗證或部署時必須調(diào)用,通常配合torch.no_grad()使用,以提高效率并避免訓練時的隨機性,本文給大家介紹PyTorch 中model.eval()推理模式,感興趣的朋友一起看看吧2025-07-07
這篇文章主要介紹了python opencv圖像混合算術(shù)運算的相關(guān)知識，本文通過實例圖文相結(jié)合給大家介紹的非常詳細，需要的朋友可以參考下2021-09-09