Python 中如何读取大文件？

8G 文件，机器只有 4G 内存，你要是上来就 read()，程序不崩才奇怪。

这类问题我一般不先看 Python 版本，也不先看什么“性能优化技巧”。先看代码里有没有这几行：

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

或者更隐蔽一点：

lines = open("access.log", encoding="utf-8").readlines()

这俩写法看着清爽，线上最容易出事。read() 是把整个文件一次性塞进内存，8G 文件不代表只占 8G 内存，字符串对象、解码、中间列表都会继续吃内存。4G 机器跑这个，基本就是等着被系统干掉。

大文件读取，第一条规矩：不要把文件当成一个整体读，要把它当成一条流。

比如处理日志，按行扫：

def scan_error_log(file_path: str):
    hit = 0
    with open(file_path, "r", encoding="utf-8", errors="replace") as f:
        for line_no, line in enumerate(f, 1):
            if "ERROR" not in line:
                continue
            hit += 1
            if hit <= 10:
                print(f"[hit] line={line_no}, text={line.rstrip()}")
    print(f"error_count={hit}")

这段代码关键不在 for line in f 多优雅，而是它不会把所有行攒起来。读一行，处理一行，这行处理完就可以被回收。

我见过不少人写成这样：

errors = []
with open("access.log", encoding="utf-8") as f:
    for line in f:
        if "ERROR" in line:
            errors.append(line)

如果错误日志特别多，这个 errors 列表还是会把内存撑爆。大文件处理时，不只是读取要流式，结果也别乱攒。该写文件就写文件，该入库就分批入库。

比如把异常行抽出来：

def dump_error_lines(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 in rf:
            if "ERROR" in line or "Traceback" in line:
                wf.write(line)

这才是正常的现场写法。别先全部读出来，再过滤，再写回去。数据量一大，中间任何一步都可能炸。

如果文件不是按行处理，比如图片、压缩包、二进制数据，那就按块读。块大小别太小，太小系统调用多；也别拍脑袋搞几百 MB，没必要。

def copy_big_file(src: str, dst: str, block_size: int = 1024 * 1024 * 8):
    total = 0
    with open(src, "rb") as rf, open(dst, "wb") as wf:
        while True:
            chunk = rf.read(block_size)
            if not chunk:
                break
            wf.write(chunk)
            total += len(chunk)
            if total % (1024 * 1024 * 512) < block_size:
                print(f"copied={total // 1024 // 1024}MB")

这里一次只读 8MB。8G 文件也就是循环多跑一会儿，内存不会因为文件大就跟着涨。

如果是 CSV，也别上来就 pandas.read_csv()。Pandas 很好，但它默认会把数据加载成 DataFrame，8G CSV 在 4G 内存机器上很可能直接翻车。

普通清洗用 csv 模块先顶住：

import csv
def count_paid_orders(csv_path: str):
    count = 0
    amount_sum = 0
    with open(csv_path, "r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row.get("status") != "PAID":
                continue
            count += 1
            amount_sum += int(row.get("amount_cent") or 0)
    print(f"paid_count={count}, amount_cent={amount_sum}")

这段代码慢不到哪里去，但稳。很多时候，稳比快重要。你连文件都读不完，谈优化没意义。

如果必须用 Pandas，也要分块：

import pandas as pd
def stat_by_chunks(csv_path: str):
    paid_count = 0
    paid_amount = 0
    for df in pd.read_csv(csv_path, chunksize=100_000):
        paid = df[df["status"] == "PAID"]
        paid_count += len(paid)
        paid_amount += paid["amount_cent"].fillna(0).astype(int).sum()
    print(paid_count, paid_amount)

chunksize 这个参数很关键，它让 Pandas 每次只吃一部分数据。别迷信某个固定值，机器内存小就调小一点，比如 5 万、10 万行一块，先跑通再说。

还有个容易被忽略的点：编码。

大文件里只要混进几行脏数据，UnicodeDecodeError 就能让任务跑到一半挂掉。离线处理我一般会加上：

errors="replace"

不是说脏数据不重要，而是先保证任务不断。后面再把异常行单独捞出来看。

最后说下 mmap。它不是把文件全部读进内存，而是做内存映射，适合随机查找、定位某些字节位置。但别一听“内存映射”就觉得它包治大文件。

import mmap
def find_keyword(file_path: str, keyword: bytes):
    with open(file_path, "rb") as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            pos = mm.find(keyword)
            if pos >= 0:
                print(f"found at byte={pos}")
            else:
                print("not found")

它适合找关键字、做二进制扫描。要是你本来就是逐行清洗日志，老老实实 for line in f 就够了。

4G 内存读 8G 文件，重点不是“怎么读进去”，而是“别读进去”。

按行读，按块读，边读边处理，结果别堆内存里。代码少一点，事故也少一点。

以上就是“Python 中如何读取大文件？”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。