处理一个大小为8G的文件，但是内存只有4G！如何实现以及需要考虑的问题？

有勇气的牛排 665 Python 2024-08-23 23:02:02

前言

灯光、灯光打过来，好，再过来一些，繁体字典准备！action

在處理大文件時，需要注意避免一次性將文件全部加載到內存中，防止內存溢出。

同時，還要關注處理效率，文件訪問模式（順序或隨機）以及最終輸出的位置。

保姆级别案例，参考下文，请尽情的复制粘贴！（忘了记得来翻博客：有勇气的牛排）

1 逐行处理

open（打开文件对象）

打開文件函數 open ，僅僅返回一個文件對象。文件內容並沒有讀取的內存中。

逐行讀取：

在使用 for line in file 或 file.readline() 的時候，文件才會被逐行讀取到內存中。

這種方式只會講當前處理的行加載到內存，而不是整個文件。

def process(line):
    print(line)

# open是打開文件對象，不是加載文件數據
with open('data.txt', 'r') as file:
    for line in file:
        # 逐行读取文件，并对每一行调用 `process` 函数
        process(line.strip())

python逐行处理数据

2 分块处理

將文件按照固定大小進行

注意：

确保数据不跨行，以免数据截断，如果一定要读取，那就将本次未处理完的数据，拼接到下一次的数据前方。

def process_chunk(chunk):
    lines = chunk.split("\r\n")
    for line in lines:
        print(line)

with open('data.txt', 'rb') as f:
    chunk_size = 1024 * 1024  # 1 MB
    while True:
        chunk = f.read(chunk_size)
        if not chunk:
            break
        process_chunk(chunk.decode())

3 使用生成器处理数据（分块）

定义：生成器是一种延迟计算的方式，合适处理大文件和大数据集。

def read_in_chunks(file_object, chunk_size=1024):
    while True:
        data = file_object.read(chunk_size)
        if not data:
            break
        yield data


with open('data.txt', 'rb') as f:
    for chunk in read_in_chunks(f):
        chunk = chunk.decode()
        lines = chunk.split("\r\n")
        for line in lines:
            print(line)

4 内存映射文件

定义：使用 mmap 将文件映射到内存中，按需读取文件内容。

4.1 方案一

mport mmap

with open('data.txt', 'r+b') as f:
    mmapped_file = mmap.mmap(f.fileno(), 0)
    for line in iter(mmapped_file.readline, b""):
        print(line.decode().replace('\r\n', ''))
    mmapped_file.close()

优点：简洁、比方案2稍微占内存

适合逐行处理的场景

4.1 方案二（推荐）

from mmap import mmap  # 从 mmap 模块导入 mmap 类，用于内存映射文件

def get_lines(filename):
    # 定义一个生成器函数 get_lines，用于按行读取指定文件
    with open(filename, 'r+') as f:
        # 以读写模式 ('r+') 打开文件，并将文件对象绑定到变量 f 上
        m = mmap(f.fileno(), 0)
        # 使用文件描述符创建内存映射对象 m。0 表示映射整个文件
        tmp = 0
        # 初始化 tmp 变量，用于记录每一行的开始位置
        for i, char in enumerate(m):
            # 使用枚举 (enumerate) 遍历内存映射对象 m 中的每一个字符
            if char == b'\n':
                # 如果当前字符是换行符 (b'\n')，则表示找到了一行的结束
                yield m[tmp:i + 1]
                # 使用 yield 生成从 tmp 到当前索引 i + 1 的字节串，表示一行的内容
                tmp = i + 1
                # 更新 tmp 为当前换行符的位置的下一位置，以便处理下一行
        # 结束文件映射处理

if __name__ == '__main__':
    # 确保以下代码块仅在直接运行脚本时执行，而不是在被导入时执行
    for line in get_lines('data.txt'):
        # 调用 get_lines 函数，逐行读取 'data.txt' 文件
        line = line.decode().replace('\r\n', '')
        # 将字节串解码为字符串，并替换掉可能存在的 '\r\n'（Windows 风格的换行符），以统一为 '\n'
        print(line)
        # 输出每一行内容

优点：

生成器模式，按需读取数据，可自定义控制数据结尾符号

适合自定义切割数据的场景

<h2><a id="_0"></a>前言</h2> 灯光、灯光打过来，好，再过来一些，繁体字典准备！action 在處理大文件時，需要注意避免一次性將文件全部加載到內存中，防止內存溢出。 同時，還要關注處理效率，文件訪問模式（順序或隨機）以及最終輸出的位置。 保姆级别案例，参考下文，请尽情的复制粘贴！（忘了记得来翻博客：有勇气的牛排） <h2><a id="1__12"></a>1 逐行处理</h2> open（打开文件对象） 打開文件函數 <code>open</code> ，僅僅返回一個文件對象。文件內容並沒有讀取的內存中。 逐行讀取： 在使用 <code>for line in file</code> 或 <code>file.readline()</code> 的時候，文件才會被逐行讀取到內存中。 這種方式只會講當前處理的行加載到內存，而不是整個文件。 <pre><div class="hljs"><code class="lang-python">def process(line): print(line) # open是打開文件對象，不是加載文件數據 with open('data.txt', 'r') as file: for line in file: # 逐行读取文件，并对每一行调用 `process` 函数 process(line.strip()) </code></div></pre> <img src="https://static.couragesteak.com/article/998e51b28fd2e024dcb84e3f3379d95e.png" alt="python逐行处理数据" /> <h2><a id="2__39"></a>2 分块处理</h2> 將文件按照固定大小進行 注意： 确保数据不跨行，以免数据截断，如果一定要读取，那就将本次未处理完的数据，拼接到下一次的数据前方。 <pre><div class="hljs"><code class="lang-python">def process_chunk(chunk): lines = chunk.split("\r\n") for line in lines: print(line) with open('data.txt', 'rb') as f: chunk_size = 1024 * 1024 # 1 MB while True: chunk = f.read(chunk_size) if not chunk: break process_chunk(chunk.decode()) </code></div></pre> <h2><a id="3__64"></a>3 使用生成器处理数据（分块）</h2> 定义：生成器是一种延迟计算的方式，合适处理大文件和大数据集。 <pre><div class="hljs"><code class="lang-python">def read_in_chunks(file_object, chunk_size=1024): while True: data = file_object.read(chunk_size) if not data: break yield data with open('data.txt', 'rb') as f: for chunk in read_in_chunks(f): chunk = chunk.decode() lines = chunk.split("\r\n") for line in lines: print(line) </code></div></pre> <h2><a id="4__87"></a>4 内存映射文件</h2> 定义：使用 mmap 将文件映射到内存中，按需读取文件内容。 <h3><a id="41__91"></a>4.1 方案一</h3> <pre><div class="hljs"><code class="lang-python">mport mmap with open('data.txt', 'r+b') as f: mmapped_file = mmap.mmap(f.fileno(), 0) for line in iter(mmapped_file.readline, b""): print(line.decode().replace('\r\n', '')) mmapped_file.close() </code></div></pre> 优点：简洁、比方案2稍微占内存 适合逐行处理的场景 <h3><a id="41___107"></a>4.1 方案二（推荐）</h3> <pre><div class="hljs"><code class="lang-python">from mmap import mmap # 从 mmap 模块导入 mmap 类，用于内存映射文件 def get_lines(filename): # 定义一个生成器函数 get_lines，用于按行读取指定文件 with open(filename, 'r+') as f: # 以读写模式 ('r+') 打开文件，并将文件对象绑定到变量 f 上 m = mmap(f.fileno(), 0) # 使用文件描述符创建内存映射对象 m。0 表示映射整个文件 tmp = 0 # 初始化 tmp 变量，用于记录每一行的开始位置 for i, char in enumerate(m): # 使用枚举 (enumerate) 遍历内存映射对象 m 中的每一个字符 if char == b'\n': # 如果当前字符是换行符 (b'\n')，则表示找到了一行的结束 yield m[tmp:i + 1] # 使用 yield 生成从 tmp 到当前索引 i + 1 的字节串，表示一行的内容 tmp = i + 1 # 更新 tmp 为当前换行符的位置的下一位置，以便处理下一行 # 结束文件映射处理 if __name__ == '__main__': # 确保以下代码块仅在直接运行脚本时执行，而不是在被导入时执行 for line in get_lines('data.txt'): # 调用 get_lines 函数，逐行读取 'data.txt' 文件 line = line.decode().replace('\r\n', '') # 将字节串解码为字符串，并替换掉可能存在的 '\r\n'（Windows 风格的换行符），以统一为 '\n' print(line) # 输出每一行内容 </code></div></pre> 优点： 生成器模式，按需读取数据，可自定义控制数据结尾符号 适合自定义切割数据的场景