Loading Academy Hub...
Process 50GB+ log files and raw JSON streams with constant O(1) memory overhead using Python yield generators, itertools, and context managers.
[!IMPORTANT] Production Context: When an ETL worker with 4GB RAM attempts
with open('events.csv') as f: lines = f.readlines(), reading a 12GB file crashes the container withMemoryErroror Linux OS SIGKILL (137). Generators yield records one at a time, keeping RAM consumption under 20MB regardless of file size.
Standard Python functions execute from entry to return and destroy their stack frames. A generator function pauses execution at yield, serializes its local frame pointer, and waits for next().
Standard Function:
Read All (10GB) ---> Put in List in RAM (10GB) ---> CRASH!
Generator Stream:
Read Line 1 (1KB) -> Yield -> Consumer Processes -> Discard from RAM
Read Line 2 (1KB) -> Yield -> Consumer Processes -> Discard from RAM
(Constant Memory Footprint: ~15MB)
Chain small generator functions to build composable, zero-memory data transformation pipelines.
json.load(f) or f.readlines() materializes the entire file into RAM (~3x raw file size!). A 20GB NDJSON dump immediately kills an 8GB container with Exit 137.yield) + itertools.islice(stream, 10000) (batch chunking) processes infinite telemetry streams in O(batch_size) RAM cleanly, feeding Arrow/Parquet row groups directly.with): Always pair file/HTTP streams with with open(...) as f: or contextlib.ExitStack() so file descriptors and S3 HTTP sockets close deterministically even mid-exception.Review the production code, company interview Q&As, incident traces, and comprehension quiz inside this lesson.
import csv
import json
from typing import Generator, Dict, Any
def stream_log_lines(file_path: str) -> Generator[str, None, None]:
"""Yields non-empty log lines one by one without reading whole file into memory."""
with open(file_path, mode='r', encoding='utf-8', buffering=1024*1024) as f:
for line in f:
stripped = line.strip()
if stripped:
yield stripped
def parse_json_records(lines: Generator[str, None, None]) -> Generator[Dict[str, Any], None, None]:
"""Parses JSON lines and yields dictionaries, filtering out malformed records."""
for line in lines:
try:
yield json.loads(line)
except json.JSONDecodeError:
continue
def filter_high_value_transactions(records: Generator[Dict[str, Any], None, None], threshold: float = 1000.0):
"""Filters transactions on the fly."""
for r in records:
if r.get('amount', 0.0) >= threshold:
yield r
# Pipeline Execution: Memory stays under 20MB for a 50GB file!
# pipeline = filter_high_value_transactions(parse_json_records(stream_log_lines('transactions_50gb.log')))
# for item in pipeline:
# process_transaction(item)MemoryError: Unable to allocate array with shape (50000000,) and data type object