如何用Python处理大文件？-Python教程-PHP中文网

处理大文件的核心是避免一次性加载，采用逐行或分块读取，利用迭代器、生成器、pandas分块和mmap等方法实现流式处理，确保内存可控。

如何用python处理大文件？

在Python中处理大文件，最核心的思路就是“不要一次性把所有数据都加载到内存里”。无论是文本文件、日志还是大型数据集，我们都需要采用流式处理或分块处理的策略，避免内存溢出，确保程序稳定高效地运行。简单来说，就是一点一点地读，一点一点地处理，而不是一口气吃成个胖子。

解决方案

处理大文件的关键在于避免将整个文件内容一次性读入内存。Python提供了多种机制来实现这一点，其中最常用且高效的方式是利用迭代器和生成器。

一个直接的方法是逐行读取文件。Python的文件对象本身就是可迭代的，这意味着你可以直接在

for

登录后复制

循环中使用它来逐行处理文件，而不需要担心内存问题。

def process_large_text_file_line_by_line(filepath):
    processed_count = 0
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            # 在这里对每一行进行处理
            # 比如清洗数据、提取信息、写入新文件等
            # print(f"Processing line: {line.strip()}")
            processed_count += 1
            if processed_count % 100000 == 0:
                print(f"Processed {processed_count} lines...")
    print(f"Finished processing {processed_count} lines.")

# 示例调用
# process_large_text_file_line_by_line('your_large_file.txt')

登录后复制

对于二进制文件或者需要固定大小块处理的情况，我们可以使用

read()

登录后复制

方法并指定每次读取的字节数，配合循环来实现分块读取。

立即学习“Python免费学习笔记（深入）”；

def process_large_binary_file_in_chunks(filepath, chunk_size=4096): # 4KB chunks
    processed_bytes = 0
    with open(filepath, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break # 文件读取完毕
            # 在这里处理读取到的二进制块
            # print(f"Processing chunk of size: {len(chunk)} bytes")
            processed_bytes += len(chunk)
            # 假设你可能需要将这些块写入另一个文件
            # output_file.write(chunk)
    print(f"Finished processing {processed_bytes} bytes.")

# 示例调用
# process_large_binary_file_in_chunks('your_large_binary_file.bin')

登录后复制

这些方法的核心思想都是将大文件“切片”处理，无论是逻辑上的行，还是物理上的字节块。这样做的好处是内存占用始终保持在一个可控的水平，不会随着文件大小的增长而线性增加。

为什么直接将大文件一次性读入内存是危险的？

这其实是个很直观的问题，但很多人在初学时容易忽略。当你用

file.read()

登录后复制

或者

file.readlines()

登录后复制

这样的方法去读取一个非常大的文件时，Python解释器会尝试把整个文件的内容，或者所有的行，一股脑地加载到你的计算机内存（RAM）中。如果文件只有几十MB，那通常没什么问题。但如果文件是几个GB，几十GB，甚至上百GB，你的内存很可能就不够用了。

想象一下，你的电脑只有8GB内存，而你却想加载一个10GB的文件。结果就是，程序会抛出

MemoryError

登录后复制

异常，或者更糟糕的是，系统会开始大量使用虚拟内存（硬盘作为内存的扩展），导致程序运行极其缓慢，甚至整个系统都卡死。这不仅仅是程序崩溃的问题，它还会严重影响用户体验和系统稳定性。我个人就遇到过因为不小心用

read()

登录后复制

读取了TB级别的日志文件，导致服务器瞬间“宕机”的尴尬经历，那次教训记忆犹新。所以，理解内存限制和避免这种“一口吃个胖子”的行为，是处理大文件最基本也最重要的常识。

Python处理大文件的核心策略：流式与生成器

面对大文件，Python的核心策略就是“流式处理”和巧妙利用“生成器”。这两种方式都是为了避免一次性加载全部数据到内存，而是按需、逐步地处理数据。

流式处理，顾名思义，就像水流一样，数据一点点地流过你的程序，你处理完一部分就放掉，再处理下一部分。最典型的就是文件对象的迭代器特性。当我们写

for line in file_object:

登录后复制

时，Python并不会一次性把所有行都读出来，而是每次迭代时才从文件中读取一行，处理完当前行后，内存中就只保留了这一行的数据，而不是整个文件。这种方式对内存极其友好。

文心大模型

百度飞桨-文心大模型 ERNIE 3.0 文本理解与创作

查看详情

生成器（Generators） 则是Python中实现流式处理的利器。它允许你定义一个函数，这个函数在每次被调用时不是返回一个值然后结束，而是“暂停”执行并“yield”一个值，下次被调用时再从上次暂停的地方继续执行。这使得生成器非常适合处理无限序列或大型数据集，因为它只在需要时才生成数据。

举个例子，假设我们有一个巨大的CSV文件，我们想筛选出某些行：

import csv

def filter_large_csv(filepath, condition_func):
    """
    一个生成器函数，用于从大型CSV文件中筛选行。
    condition_func 是一个接受一行数据（字典）并返回True/False的函数。
    """
    with open(filepath, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f) # 使用DictReader方便按列名访问
        for row in reader:
            if condition_func(row):
                yield row # 满足条件的行才会被生成

# 假设我们有一个名为 'large_data.csv' 的文件
# 并且我们想筛选 'age' 列大于30的行
def is_older_than_30(row):
    try:
        return int(row.get('age', 0)) > 30
    except ValueError:
        return False # 处理非数字的age值

# 实际使用时：
# for filtered_row in filter_large_csv('large_data.csv', is_older_than_30):
#     print(filtered_row)
#     # 在这里对筛选出的行进行进一步处理
#     # 比如写入新的CSV文件，或者进行统计分析

登录后复制

这个

filter_large_csv

登录后复制

函数就是一个生成器。它不会把所有符合条件的行都收集到一个列表里再返回，而是每找到一行符合条件的就

yield

登录后复制

出来。这样，无论原文件有多大，只要你不是把所有

yield

登录后复制

出来的结果又全部存到一个列表里，内存占用就能保持在很低的水平。这种模式在处理日志、网络流或者任何无法一次性加载的数据源时都非常有效。

处理大型CSV或日志文件时有哪些高效方法？

对于大型CSV或日志文件，除了基本的逐行读取和生成器，我们还可以结合一些专门的库和技巧来提升效率和便利性。

1. Pandas分块读取CSV： 如果你处理的是结构化的CSV文件，并且需要进行数据分析，

pandas

登录后复制

是一个非常强大的工具。虽然

pandas.read_csv()

登录后复制

默认会尝试加载整个文件，但它提供了一个

chunksize

登录后复制

参数，允许你分块读取数据。

import pandas as pd

def process_large_csv_with_pandas_chunks(filepath, chunk_size=10000):
    """
    使用pandas分块读取并处理大型CSV文件。
    """
    total_rows_processed = 0
    # iterator=True 返回一个TextFileReader对象，可以迭代
    for chunk_df in pd.read_csv(filepath, chunksize=chunk_size):
        # chunk_df 是一个DataFrame，包含chunk_size行数据
        # 在这里对每个数据块进行处理
        # 例如，计算每块的平均值，筛选数据，或者写入数据库
        print(f"Processing a chunk of {len(chunk_df)} rows. Current total: {total_rows_processed + len(chunk_df)}")
        # 示例：计算某个列的平均值
        # if 'value_column' in chunk_df.columns:
        #     avg_value = chunk_df['value_column'].mean()
        #     print(f"Average value in this chunk: {avg_value}")

        total_rows_processed += len(chunk_df)
    print(f"Finished processing {total_rows_processed} rows in total.")

# 示例调用
# process_large_csv_with_pandas_chunks('your_large_data.csv')

登录后复制

chunksize

登录后复制

的魔力在于，它让你可以像处理普通DataFrame一样操作每个数据块，同时又避免了内存问题。你可以对每个

chunk_df

登录后复制

执行聚合、过滤、转换等操作，然后将结果累积起来。

2. 针对日志文件的模式匹配与状态机： 日志文件通常是非结构化的，或者半结构化的。除了逐行读取，我们经常需要结合正则表达式来提取关键信息。如果日志条目跨越多行，你可能还需要构建一个简单的状态机来识别和组合完整的日志事件。

import re

def parse_large_log_file(filepath, pattern):
    """
    逐行读取大型日志文件，并使用正则表达式提取匹配的内容。
    """
    compiled_pattern = re.compile(pattern)
    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
        for line_num, line in enumerate(f, 1):
            match = compiled_pattern.search(line)
            if match:
                # 假设你只想提取匹配到的所有捕获组
                yield (line_num, match.groups())
            # 也可以在这里处理没有匹配到的行，或者根据业务逻辑进行其他操作

# 示例：提取Apache日志中的IP地址和请求路径
# log_pattern = r'^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - - \[.*?\] "GET (.+?) HTTP'
# for line_num, data in parse_large_log_file('access.log', log_pattern):
#     ip_address, request_path = data
#     # print(f"Line {line_num}: IP={ip_address}, Path={request_path}")

登录后复制

对于更复杂的日志结构，比如一个日志事件可能包含多行（例如，Java堆栈跟踪），你可以维护一个缓冲区，当识别到事件的开始和结束标记时，就处理缓冲区中的所有行。这本质上就是一种简单的状态机模式，确保你不会漏掉任何一个完整的日志事件。

3. 使用

mmap

登录后复制

进行内存映射： 对于某些特定的场景，特别是需要随机访问大文件中的数据，或者文件非常大但你只需要处理其中一小部分时，

mmap

登录后复制

模块是一个高级选项。它将文件内容映射到进程的虚拟内存空间，让你能够像访问内存数组一样访问文件，而操作系统会负责按需加载文件页面。

import mmap
import os

def search_in_large_file_with_mmap(filepath, search_term):
    """
    使用mmap在大型文件中查找特定字符串。
    """
    with open(filepath, 'r+b') as f: # 'r+b' 允许读写二进制模式
        # mmap.mmap(fileno, length, access=ACCESS_READ)
        # length=0 意味着映射整个文件
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            # 搜索是字节串，所以search_term需要编码
            index = mm.find(search_term.encode('utf-8'))
            if index != -1:
                print(f"Found '{search_term}' at byte offset: {index}")
                # 你可以进一步读取该位置附近的数据
                # 例如：print(mm[index:index+50].decode('utf-8'))
            else:
                print(f"'{search_term}' not found.")

# 示例调用
# if os.path.exists('your_large_text_file.txt'):
#     search_in_large_file_with_mmap('your_large_text_file.txt', 'important_keyword')

登录后复制