Python中大文件读写的两大优化策略详解

with open("large_log.txt", "r") as f:
    content = f.read()  # 一次性加载整个文件到内存

def chunked_read(filename, chunk_size=65536):
    """
    按块读取大文件，适用于文本或二进制文件。
    
    :param filename: 文件路径
    :param chunk_size: 每次读取的字节数，默认 64KB
    :yield: 每一块的数据
    """
    with open(filename, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 使用示例
if __name__ == "__main__":
    file_path = "large_file.log"
    total_bytes = 0
    
    for i, chunk in enumerate(chunked_read(file_path, chunk_size=65536)):
        total_bytes += len(chunk)
        print(f"Chunk {i+1}: {len(chunk)} bytes read")
        
        # 模拟处理逻辑（如解析日志、压缩等）
        # process_chunk(chunk)

    print(f"✅ Total size: {total_bytes} bytes")

def chunked_write(output_filename, chunks, chunk_size=65536):
    """
    将数据块写入文件，支持流式输出。
    
    :param output_filename: 输出文件名
    :param chunks: 可迭代的块数据（字节串）
    :param chunk_size: 块大小（可选）
    """
    with open(output_filename, 'wb') as f:
        for chunk in chunks:
            if isinstance(chunk, str):
                chunk = chunk.encode('utf-8')
            f.write(chunk)
        print(f"💾 Successfully wrote to {output_filename}")

# 模拟生成数据块
def generate_data_chunks():
    for i in range(1000):
        yield f"Line {i}: This is a test line.\n".encode('utf-8')

# 执行写入
chunked_write("output_large_file.txt", generate_data_chunks())

import re
from datetime import datetime

# 正则表达式匹配错误日志
ERROR_PATTERN = re.compile(r'\[ERROR\]')
DATE_PATTERN = re.compile(r'(\d{4}-\d{2}-\d{2})')

def filter_and_split_logs(input_file, output_dir, chunk_size=65536):
    """
    分块读取日志，提取 ERROR 并按日期分组输出。
    """
    current_date = None
    buffer = b""  # 用于跨块的缓冲区
    file_handles = {}

    def get_output_handle(date_str):
        """获取对应日期的输出文件句柄"""
        if date_str not in file_handles:
            filename = f"{output_dir}/error_{date_str}.log"
            file_handles[date_str] = open(filename, 'ab')
        return file_handles[date_str]

    with open(input_file, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break

            # 合并缓冲区与新块
            combined = buffer + chunk
            buffer = b""  # 清空缓冲区

            # 查找最后一个换行符位置，防止截断行
            last_newline = combined.rfind(b'\n')
            if last_newline == -1:
                # 整个块都没有换行，全部放入缓冲区
                buffer = combined
                continue
            else:
                # 分割为已完整行 + 未完成行
                complete_lines = combined[:last_newline + 1]
                buffer = combined[last_newline + 1:]

            # 处理完整行
            for line in complete_lines.splitlines(True):  # True 保留换行符
                line = line.strip()
                if not line:
                    continue

                # 检查是否包含 [ERROR]
                if ERROR_PATTERN.search(line.decode('utf-8', errors='ignore')):
                    # 提取日期
                    match = DATE_PATTERN.search(line.decode('utf-8', errors='ignore'))
                    if match:
                        date_str = match.group(1)
                    else:
                        date_str = "unknown"

                    # 写入对应文件
                    handle = get_output_handle(date_str)
                    handle.write(line + b'\n')

    # 关闭所有打开的文件
    for handle in file_handles.values():
        handle.close()

    print(f"✅ Processing completed. Errors saved in {output_dir}/")

# 调用函数
filter_and_split_logs("server_logs_2024.log", "./errors_by_date")

import asyncio
import aiofiles

async def async_chunked_read(filename, chunk_size=65536):
    """
    异步分块读取文件。
    """
    async with aiofiles.open(filename, 'rb') as f:
        while True:
            chunk = await f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 异步主函数
async def main():
    file_path = "large_file.bin"
    async for chunk in async_chunked_read(file_path):
        # 模拟耗时处理（如哈希计算、压缩）
        await asyncio.sleep(0.001)  # 模拟延迟
        print(f"Processed {len(chunk)} bytes")

# 运行异步任务
if __name__ == "__main__":
    asyncio.run(main())

import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def robust_chunked_process(input_file, output_file, chunk_size=65536, max_retries=3):
    """
    带重试机制的稳健分块处理。
    """
    if not os.path.exists(input_file):
        raise FileNotFoundError(f"Input file not found: {input_file}")

    retry_count = 0
    while retry_count < max_retries:
        try:
            with open(input_file, 'rb') as fin, open(output_file, 'wb') as fout:
                while True:
                    chunk = fin.read(chunk_size)
                    if not chunk:
                        break
                    fout.write(chunk)
            logger.info(f"✅ Successfully processed {input_file} -> {output_file}")
            return True

        except (IOError, OSError) as e:
            retry_count += 1
            logger.warning(f"IO error on attempt {retry_count}: {e}")
            if retry_count >= max_retries:
                logger.error(f"❌ Failed after {max_retries} retries.")
                raise
            else:
                # 简单延时重试
                import time
                time.sleep(1)

    return False

class LargeFileProcessor:
    def __init__(self, chunk_size=65536):
        self.chunk_size = chunk_size

    def read_in_chunks(self, filename):
        with open(filename, 'rb') as f:
            while True:
                chunk = f.read(self.chunk_size)
                if not chunk:
                    break
                yield chunk

    def write_from_chunks(self, filename, chunks):
        with open(filename, 'wb') as f:
            for chunk in chunks:
                f.write(chunk)

    def copy_with_progress(self, src, dst, show_progress=True):
        total = 0
        copied = 0

        # 获取总大小（可选）
        try:
            total = os.path.getsize(src)
        except OSError:
            pass

        with open(src, 'rb') as fin, open(dst, 'wb') as fout:
            while True:
                chunk = fin.read(self.chunk_size)
                if not chunk:
                    break
                fout.write(chunk)
                copied += len(chunk)
                if show_progress and total > 0:
                    percent = (copied / total) * 100
                    print(f"\rProgress: {percent:.1f}% ({copied}/{total} bytes)", end="", flush=True)
        if show_progress:
            print("\n✅ Copy complete.")

# 使用示例
processor = LargeFileProcessor(chunk_size=131072)  # 128KB
processor.copy_with_progress("big_file.zip", "backup.zip")

import time
import os

def benchmark_chunk_size(filename, sizes=[8192, 65536, 262144, 1048576]):
    """
    测试不同块大小下的读取性能。
    """
    for size in sizes:
        start_time = time.time()
        total_bytes = 0
        with open(filename, 'rb') as f:
            while True:
                chunk = f.read(size)
                if not chunk:
                    break
                total_bytes += len(chunk)
        duration = time.time() - start_time
        print(f"Chunk size: {size} bytes → Time: {duration:.3f}s, Speed: {total_bytes/duration/1e6:.2f} MB/s")

# 生成测试文件（仅一次）
if __name__ == "__main__":
    test_file = "test_large_file.dat"
    if not os.path.exists(test_file):
        with open(test_file, 'wb') as f:
            f.write(b"Hello World!" * 1000000)  # ~10MB
    benchmark_chunk_size(test_file)