Python 中如何读取大文件，例如内存只有 4G，如何读取一个大小为 8G 的文件

8G 文件一上来就 read()，4G 内存的机器基本不用挣扎，进程会直接把自己送走。

这种问题我一般不先看 Python 版本，也不先怀疑磁盘。第一眼先找这类代码：

with open("access.log", "r", encoding="utf-8") as f:
    data = f.read()
for line in data.splitlines():
    handle(line)

这段代码看着很正常，问题也很直接：它把 8G 文件一次性塞进内存了。

更麻烦的是，8G 文件不等于只占 8G 内存。文本解码、字符串对象、splitlines() 再复制一份，内存涨得比想象快。4G 内存的机器跑这种代码，慢一点是运气好，直接 OOM 才是常态。

我处理大文件，一般先换成流式读取。不要把文件当成一个整体，文件在磁盘上可以是 8G、80G，但程序每次只拿一小段。

最常见的是按行读。

defhandle_line(line: str):
# 这里只做一行的处理，不要把结果都塞到 list 里
if"ERROR"notin line:
return
# 假设线上只关心错误日志里的订单号
    parts = line.strip().split("|")
if len(parts) >= 4:
        print(parts[1], parts[3])
defscan_log(path: str):
with open(path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
            handle_line(line)
scan_log("/data/logs/order-2026-06-01.log")

这里有个细节，for line in f 不是把文件一次性读完。它底层会借助缓冲区一点点读，内存里只留当前这一行和少量缓冲数据。

这种写法我在线上用得最多，尤其是日志清洗、账单核对、导入前校验这些活。别一上来就搞花的，先把内存稳住。

但按行读也不是万能的。

有些文件不是标准文本，比如压缩前的二进制包、图片归档、模型文件，或者一行就几百 MB 的 JSON。这时候按行读就不合适了，要按块读。

from pathlib import Path
defcopy_big_file(src: str, dst: str, block_size: int = 4 * 1024 * 1024):
    src_path = Path(src)
    copied = 0
with open(src_path, "rb") as rf, open(dst, "wb") as wf:
whileTrue:
            chunk = rf.read(block_size)
ifnot chunk:
break
            wf.write(chunk)
            copied += len(chunk)
if copied % (512 * 1024 * 1024) < block_size:
                print(f"copied={copied / 1024 / 1024:.0f}MB")
copy_big_file("/data/raw/big.dat", "/data/bak/big.dat")

我这里一般不会把 block_size 调得特别大。4MB、8MB 都够用。调到 512MB 没什么意义，反而让内存抖得厉害。大文件处理最怕的不是慢一点，是处理到一半机器开始换页，磁盘灯狂闪，程序像死了一样。

还有一种更常见的业务场景：8G CSV，要统计某个字段，或者筛出异常数据。

这种最容易写成这样：

rows = list(csv.reader(f))

这行在小文件里没毛病，在 8G 文件里就是事故入口。

我会写成边读边处理，结果也别全堆内存里，能落盘就落盘，能聚合就聚合。

import csv
from collections import defaultdict
defstat_pay_amount(path: str):
    amount_by_day = defaultdict(int)
with open(path, "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
for idx, row in enumerate(reader, 1):
            day = row.get("pay_day")
            amount = row.get("amount")
ifnot day ornot amount:
continue
try:
                amount_by_day[day] += int(amount)
except ValueError:
                print(f"bad amount at line={idx}, value={amount!r}")
if idx % 1_000_000 == 0:
                print(f"checked lines={idx}")
return amount_by_day
result = stat_pay_amount("/data/bill/pay_202606.csv")
for day, total in sorted(result.items()):
    print(day, total)

这段代码有两个我比较在意的点。

第一，结果集是按天聚合的，不是把所有明细都放进列表。哪怕原文件 8G，最后内存里可能就几十个 key。

第二，隔一段打一次进度。处理大文件没有进度日志很难受，尤其跑了半小时以后，你不知道它是卡住了，还是还在正常干活。

如果是要从 8G 文件里筛出一部分数据，别返回一个大列表，直接写到新文件。

defpick_failed_orders(src: str, dst: str):
with open(src, "r", encoding="utf-8", errors="replace") as rf, \
         open(dst, "w", encoding="utf-8") as wf:
for line_no, line in enumerate(rf, 1):
if"|PAY_FAILED|"notin line:
continue
            wf.write(line)
if line_no % 2_000_000 == 0:
                print(f"scan line={line_no}")
pick_failed_orders(
"/data/logs/pay.log",
"/data/tmp/pay_failed.log"
)

我见过不少代码，前面读文件已经改成流式了，后面又来一句：

matched.append(line)

然后文件里刚好有几千万条命中数据，内存照样炸。

所以大文件处理要盯住两处：入口不要一次性读，出口不要无限制攒。

还有人会问 mmap 能不能用。

能用，但别神化。mmap 是把文件映射到虚拟内存，适合随机访问，比如你要在一个大文件里按偏移量查一段内容。它不是让 4G 内存凭空吃下 8G 文件。

比如只检查文件头尾，可以这样：

import mmap
defread_head_tail(path: str):
with open(path, "rb") as f:
with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            head = mm[:128]
            tail = mm[-128:]
            print("head:", head)
            print("tail:", tail)

但如果只是从头扫到尾，普通缓冲读取就够了。不要为了显得高级，把事情搞复杂。

最后留一个我排查时常用的小习惯。跑大文件脚本前，先把明显危险的写法搜掉：

grep -R "read()" ./scripts
grep -R "readlines()" ./scripts
grep -R "list(" ./scripts

不是说这些一定不能用，而是在大文件脚本里看到它们，我会多看两眼。

8G 文件、4G 内存，这个题没什么玄学。别把文件一次性搬进内存，按行读、按块读、边读边处理、边处理边落盘。

代码稳不稳，很多时候就差这一点。