Python'da Yüksek Performanslı Akış İşleme

Büyük ölçekli log dosyalarının işlenmesi, modern veri mühendisliğinde yaygın bir zorluktur. Üretim ortamı sorunlarını halletmek, kullanıcı davranışlarını analiz etmek veya verileri bir makine öğrenimi boru hattına beslemek olsun, mevcut sistem RAM'ini aşan dosyaları işleme yeteneği hayati önem taşır. Standart dosya I/O yöntemleriyle tüm dosyayı belleğe yükleyen geleneksel yaklaşımlar, gigabaytlarca veriyle çalışırken bellek tükenmesine (OutOfMemory hataları) veya ciddi performans düşüşüne yol açabilir.

Çözüm, Python'un güçlü döngü (iterator) protokolünü benimsemekte yatmaktadır. Üreteçleri (generators) bağlam yöneticileriyle (context managers) birleştiren geliştiriciler, veriyi sırayla akıtan sağlam ve bellek verimli boru hatları oluşturabilir. Bu yaklaşım, dosya boyutu ne olursa olsun, uygulamaların sabit bir bellek iziyle devasa veri setlerini işlemesine olanak tanır. Bu yazıda, GB ölçekli loglar için yüksek performanslı akış çözümlerini nasıl mimarileştireceğimizi inceleyeceğiz.

Üreteçlerle Bellek Verimliliğini Anlamak

Standart bir dosya okuma işlemi genellikle tüm dosyayı bir liste veya dize olarak belleğe yükler. 5 GB'lık bir log dosyası için bu, 5 GB RAM ve ek yük gerektirir. Bir üreteç ise tam tersine, bir seferde bir öğe üretir. Bir dosyadan satırları okumak için kullanıldığında, üreteç sadece mevcut satırı belleğe yükler, işler, üretir ve sonraki satıra geçmeden önce onu bellekten siler.

Bu yavaş değerlendirme (lazy evaluation) stratejisi, verimli akışın temel taşıdır. Veri okuma hızını veri işleme hızından ayırarak, uygulamanın RAM yerine disk alanıyla sınırsız şekilde ölçeklenmesine olanak tanır.

Güçlü Bir Akış Boru Hattı Oluşturma

Basit döngüler üreteçleri kullanabilirken, üretim seviyesindeki kodlar uygun kaynak yönetimi gerektirir. İşte burada bağlam yöneticisi deseni vazgeçilmez hale gelir. Dosya işlemlerinin doğru şekilde açılıp kapatılmasını sağlar, hatta istisnalar oluşsa bile zamanla sistem dosya tanımlayıcılarının tükenmesine neden olabilecek kaynak sızıntılarını önler.

Aşağıda, log satırlarını işleyen, bunları parse eden ve yapılandırılmış veri üreten pratik bir akış işleyicisi uygulaması yer almaktadır:

import json
import logging
from typing import Iterator, Dict, Any

# Hata izleme için log yapılandırması
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def stream_log_file(filepath: str, chunk_size: int = 8192) -> Iterator[Dict[str, Any]]:
    """
    Büyük bir dosyadan log satırlarını akıtmak için üreteç fonksiyonu.
    Dosya işleminin her zaman kapatılmasını sağlamak için bağlam yönetimi kullanır.
    """
    try:
        with open(filepath, 'r', encoding='utf-8', buffering=chunk_size) as file:
            for line_number, line in enumerate(file, start=1):
                line = line.strip()
                if not line:
                    continue
                
                try:
                    # JSON log satırını parse et (örnek)
                    if line.startswith('{'):
                        data = json.loads(line)
                        data['line_number'] = line_number
                        yield data
                    else:
                        # Gerekirse JSON olmayan satırları işle
                        logging.warning(f"Hatalı satır atlanıyor {line_number}")
                except json.JSONDecodeError as e:
                    logging.error(f"Satır {line_number}'de JSON parse hatası: {e}")
                    continue
                    
    except FileNotFoundError:
        logging.error(f"Dosya bulunamadı: {filepath}")
        raise
    except PermissionError:
        logging.error(f"İzin reddedildi: {filepath}")
        raise

# Kullanım Örneği
if __name__ == "__main__":
    log_path = "large_server_log.json"
    processed_count = 0
    
    # Üreteç bir seferde bir öğe üretir, bellek kullanımını sabit tutar
    for record in stream_log_file(log_path):
        # Kaydı hemen işle
        if record.get('severity') == 'ERROR':
            print(f"Satır {record['line_number']}'de Kritik Hata: {record.get('message')}")
        
        processed_count += 1
        
        # Test için sadece ilk 100 kaydı işleniyorsa erken çık
        if processed_count >= 100:
            break
            
    print(f"{processed_count} kayıt minimal bellek kullanımıyla başarıyla işlendi.")