K последних строк большого файла

Решение: Чтение K последних строк из 50 ГБ файла

Описание проблемы

Нужно прочитать последние K строк из файла размером 50 ГБ при ограниченной памяти. Наивный подход (загрузить всё в память или прочитать весь файл) невозможен. Требуется оптимальная стратегия.

Ограничения:

• Размер файла: 50 ГБ
• Доступная память: ограничена (обычно несколько ГБ)
• Нужно вернуть: K строк

Подход 1: Потоковый с Deque (Оптимальный)

from collections import deque

def read_last_k_lines(filename, k):
    if k <= 0:
        return []
    
    lines_buffer = deque(maxlen=k)
    
    with open(filename, 'r', encoding='utf-8', errors='replace') as f:
        for line in f:
            lines_buffer.append(line.rstrip('\\n'))
    
    return list(lines_buffer)

Этот подход:

• Читает файл один раз
• Использует O(k × L) памяти, где L = средняя длина строки
• Автоматически выбрасывает старые строки при переполнении deque
• Простой и надёжный

Подход 2: Бинарный поиск с конца файла (Быстрый)

def read_last_k_lines_fast(filename, k):
    if k <= 0:
        return []
    
    with open(filename, 'rb') as f:
        f.seek(0, 2)
        file_size = f.tell()
        
        buffer = []
        block_size = 1024 * 1024
        offset = 0
        incomplete = b''
        
        while offset < file_size:
            read_pos = max(0, file_size - offset - block_size)
            bytes_read = min(block_size, file_size - read_pos)
            
            f.seek(read_pos)
            chunk = f.read(bytes_read) + incomplete
            
            lines = chunk.split(b'\\n')
            incomplete = lines[0]
            lines = lines[1:]
            
            buffer.extend(reversed(lines))
            
            if len(buffer) >= k:
                break
            
            offset += block_size
        
        if incomplete:
            buffer.append(incomplete)
        
        result = buffer[:k]
        result.reverse()
        return [l.decode('utf-8', errors='replace') for l in result]

Достоинства:

• Не читает весь файл
• Быстро для больших файлов
• Идеален если нужны только последние строки

Подход 3: Unix tail (Лучший в системе)

import subprocess

def read_last_k_lines_unix(filename, k):
    if k <= 0:
        return []
    
    try:
        result = subprocess.run(
            ['tail', '-n', str(k), filename],
            capture_output=True,
            text=True,
            timeout=30
        )
        if result.returncode == 0:
            return result.stdout.strip().split('\\n')
    except Exception as e:
        print(f"Ошибка: {e}")
    
    return []

Сравнение подходов

Анализ сложности

Deque:

• Временная: O(n) где n = количество строк в файле
• Пространственная: O(k × L) где L = длина строки
• Преимущество: гарантированно работает

Бинарный поиск:

• Временная: O(log(bytes) × block_size)
• Пространственная: O(k × L)
• Преимущество: может не читать весь файл

Граничные случаи

test_cases = [
    (0, []),
    (1, ['последняя строка']),
    (999999999, 'все строки'),
    (-5, []),
]

Практическое применение в QA Automation

• Анализ логов при тестировании (последние ошибки)
• Проверка результатов long-running тестов
• Обработка больших артефактов CI/CD
• Быстрое извлечение итогов из отчётов
• Отслеживание статуса нагрузочного тестирования