如何優(yōu)雅地處理超大文件？用流式 + yield，輕松應(yīng)對巨型日志！
作者：Ss肥魚
面對超大文件（幾十 GB 的日志），你還在一行一行慢慢讀？其實高手更喜歡用流式處理 + yield，不僅優(yōu)雅，還能讓內(nèi)存占用低得嚇人！
面對超大文件（幾十 GB 的日志），你還在一行一行慢慢讀？其實高手更喜歡用流式處理 + yield，不僅優(yōu)雅，還能讓內(nèi)存占用低得嚇人！
為什么不能一次性讀完？
最常見錯誤寫法是：
with open('huge.log', 'r', encoding='utf-8') as f:
data = f.read() # ?問題：
- .read() 會把整個文件內(nèi)容加載到內(nèi)存！
- 文件一旦超過內(nèi)存大小，程序直接 OOM（內(nèi)存溢出）。
所以，處理大文件的核心是：永遠不要一次性讀完！
正確做法：流式逐行讀取
最基礎(chǔ)的穩(wěn)健方案是：
def read_file_line_by_line(filepath: str):
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
yield line.rstrip('\n') # 用 yield 每次返回一行怎么用？
for line in read_file_line_by_line('huge.log'):
if 'ERROR' in line:
print(line)yield 的好處：
- 不是一次性讀全部，而是懶加載，一行一行處理。
- 保證內(nèi)存占用極低，即使是 100GB 也沒問題！
更高效的方式：按塊（chunk）讀取 + yield
逐行讀取有點慢？那可以分塊讀取，然后用 yield：
def read_file_in_chunks(filepath: str, chunk_size: int = 1024 * 1024):
with open(filepath, 'r', encoding='utf-8') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk每次讀 1MB（可以調(diào)整），然后按需處理。
注意：如果按塊讀取，可能存在行被切斷的問題，通常需要自己補全一行。
高級版本是：
def read_file_by_line_efficient(filepath: str, buffer_size: int = 1024*1024):
withopen(filepath, 'r', encoding='utf-8') as f:
buffer = ''
whileTrue:
chunk = f.read(buffer_size)
ifnot chunk:
break
buffer += chunk
while'\n'in buffer:
line, buffer = buffer.split('\n', 1)
yield line
if buffer:
yield buffer這段代碼解決了跨塊的半行問題！
進階版：加個實時進度顯示！
如果你想邊讀邊看到進度，可以這樣做：
import os
def read_file_with_progress(filepath: str):
total_size = os.path.getsize(filepath)
read_size = 0
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
read_size += len(line.encode('utf-8'))
progress = (read_size / total_size) * 100
print(f'\r讀取進度：{progress:.2f}%', end='')
yield line.rstrip('\n')每處理一行，更新一次進度，體驗感拉滿！
小結(jié)
方法 | 適合場景 | 說明 |
整體讀 | 小文件（KB/MB） | 不推薦大文件 |
逐行 | 大多數(shù)大文件場景 | 穩(wěn)定可靠 |
按塊 | 超大文件、性能要求高 | 注意跨行處理 |
加進度顯示 | 體驗更好 | 推薦實際項目用 |
責(zé)任編輯：趙寧寧
來源：
Ssoul肥魚
