Python中几种高效读取大文件的完整指南

with open('large_file.txt', 'r', encoding='utf-8') as f:
    for line in f:  # 逐行读取，内存友好
        process_line(line)  # 处理每一行

with open('large_file.txt', 'r') as f:
    while True:
        line = f.readline()
        if not line:  # 到达文件末尾
            break
        process_line(line)

BUFFER_SIZE = 1024 * 1024  # 1MB的缓冲区

with open('large_file.bin', 'rb') as f:
    while True:
        chunk = f.read(BUFFER_SIZE)
        if not chunk:  # 文件结束
            break
        process_chunk(chunk)

from functools import partial

chunk_size = 1024 * 1024  # 1MB
with open('large_file.bin', 'rb') as f:
    for chunk in iter(partial(f.read, chunk_size), b''):
        process_chunk(chunk)

import mmap

with open('large_file.bin', 'r+b') as f:
    # 创建内存映射
    mm = mmap.mmap(f.fileno(), 0)
    
    # 像操作字符串一样操作文件
    print(mm[:100])  # 读取前100字节
    
    # 可以搜索内容
    index = mm.find(b'some_pattern')
    if index != -1:
        print(f"Found at position {index}")
    
    mm.close()  # 记得关闭映射

def read_large_file(file_path, chunk_size=1024*1024):
    """生成器函数，逐块读取大文件"""
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 使用生成器
for chunk in read_large_file('huge_file.bin'):
    process_chunk(chunk)

import gzip
import shutil

with gzip.open('large_file.gz', 'rb') as f_in:
    with open('large_file_extracted', 'wb') as f_out:
        shutil.copyfileobj(f_in, f_out)  # 流式复制

import zipfile

with zipfile.ZipFile('large_file.zip', 'r') as z:
    with z.open('file_inside.zip') as f:
        for line in f:
            process_line(line)

from concurrent.futures import ThreadPoolExecutor
import os

def process_chunk(start, end, file_path):
    """处理文件的指定部分"""
    with open(file_path, 'rb') as f:
        f.seek(start)
        chunk = f.read(end - start)
        # 处理chunk...

def parallel_file_processing(file_path, num_threads=4):
    file_size = os.path.getsize(file_path)
    chunk_size = file_size // num_threads
    
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        futures = []
        for i in range(num_threads):
            start = i * chunk_size
            end = start + chunk_size if i != num_threads - 1 else file_size
            futures.append(executor.submit(process_chunk, start, end, file_path))
        
        # 等待所有任务完成
        for future in concurrent.futures.as_completed(futures):
            future.result()

import dask.dataframe as dd

# 创建延迟计算的DataFrame
df = dd.read_csv('very_large_file.csv', blocksize=25e6)  # 25MB每块

# 执行操作（惰性计算）
result = df.groupby('column').mean().compute()  # 实际计算

import tables

# 打开HDF5文件
h5file = tables.open_file('large_data.h5', mode='r')

# 访问数据集
table = h5file.root.data.table
for row in table.iterrows():  # 迭代访问
    process_row(row)

h5file.close()

import sqlite3

# 将数据导入SQLite
conn = sqlite3.connect(':memory:')  # 或磁盘数据库
cursor = conn.cursor()
cursor.execute('CREATE TABLE data (col1, col2, col3)')

# 批量插入数据
with open('large_file.csv') as f:
    # 使用生成器避免内存问题
    data_gen = (line.strip().split(',') for line in f)
    cursor.executemany('INSERT INTO data VALUES (?, ?, ?)', data_gen)

conn.commit()

import csv
from collections import namedtuple
from itertools import islice

def process_large_csv(file_path, batch_size=10000):
    """分批处理大型CSV文件"""
    
    # 定义行结构
    CSVRow = namedtuple('CSVRow', ['id', 'name', 'value'])
    
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        headers = next(reader)  # 跳过标题行
        
        while True:
            # 读取一批行
            batch = list(islice(reader, batch_size))
            if not batch:
                break  # 文件结束
                
            # 处理批次
            rows = [CSVRow(*row) for row in batch]
            process_batch(rows)
            
            # 可选：显示进度
            print(f"Processed {len(batch)} rows")

def process_batch(rows):
    """处理一批数据"""
    # 这里添加实际处理逻辑
    pass

# 使用
process_large_csv('huge_dataset.csv')