Build 3 — High-Performance Data Processor

# Build 3 — High-Performance Data Processor ## The brief Build a data processor that: - Reads JSON-lines logs from many files in parallel. - Parses and validates each record. - Applies some heavy CPU work (regex extraction, calculations). - Writes results to a single output. Use asyncio for I/O + multiprocessing for CPU. Profile before and after; achieve a 5x+ speedup over a naive single-process synchronous version. ## Step 1: Set up the problem (15 min) Generate test data: 100 JSON-lines files, each with 10K records. ```python # generate_data.py import json import random import string from pathlib import Path OUTPUT_DIR = Path("data") OUTPUT_DIR.mkdir(exist_ok=True) def random_log(): return { "timestamp": "2026-05-26T12:00:00Z", "level": random.choice(["INFO", "WARN", "ERROR"]), "user_id": random.randint(1, 100000), "message": ''.join(random.choices(string.ascii_letters + " ", k=200)), "ip": f"{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}", } for i in range(100): path = OUTPUT_DIR / f"log_{i:03d}.jsonl" with open(path, "w") as f: for _ in range(10000): f.write(json.dumps(random_log()) + "\n") ``` 100 files × 10K records = 1M records, ~200MB total. ## Step 2: Write a naive version (10 min) ```python # naive.py import json from pathlib import Path import re import time INPUT_DIR = Path("data") EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}") def process_record(record): # Simulate CPU work text = record["message"] emails = EMAIL_RE.findall(text) word_count = len(text.split()) return { "user_id": record["user_id"], "level": record["level"], "word_count": word_count, "email_count": len(emails), } def main(): start = time.time() results = [] for path in INPUT_DIR.glob("*.jsonl"): with open(path) as f: for line in f: record = json.loads(line) results.append(process_record(record)) print(f"Processed {len(results)} records in {time.time() - start:.2f}s") if __name__ == "__main__": main() ``` Measure: this should take 30-60 seconds. ## Step 3: Profile the naive version (10 min) ```bash python -m cProfile -o profile.out naive.py python -c "import pstats; pstats.Stats('profile.out').sort_stats('cumulative').print_stats(20)" ``` Or use py-spy for a flame graph: ```bash py-spy record -o flame.svg -- python naive.py ``` Identify hot spots: JSON parsing, regex, file I/O. ## Step 4: Optimized version (40 min) Plan: - File reading: parallel (async or threading). - Record processing: CPU-bound → multiprocessing. ```python # optimized.py import asyncio import json import time from concurrent.futures import ProcessPoolExecutor from pathlib import Path import re INPUT_DIR = Path("data") EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}") def process_file(path: Path) -> list[dict]: """Process one file. Runs in worker process.""" results = [] with open(path) as f: for line in f: record = json.loads(line) text = record["message"] emails = EMAIL_RE.findall(text) word_count = len(text.split()) results.append({ "user_id": record["user_id"], "level": record["level"], "word_count": word_count, "email_count": len(emails), }) return results async def main(): start = time.time() paths = list(INPUT_DIR.glob("*.jsonl")) loop = asyncio.get_event_loop() # Process files in parallel across CPU cores with ProcessPoolExecutor() as pool: tasks = [ loop.run_in_executor(pool, process_file, path) for path in paths ] all_results = await asyncio.gather(*tasks) # Flatten results = [r for batch in all_results for r in batch] print(f"Processed {len(results)} records in {time.time() - start:.2f}s") if __name__ == "__main__": asyncio.run(main()) ``` Why this works: - ProcessPoolExecutor uses N processes (default = CPU count). - Each process handles whole files independently. - asyncio.gather collects results. - File I/O happens inside the worker process (no cross-process file passing). ## Step 5: Measure improvement (10 min) ```bash # Naive time python naive.py # Optimized time python optimized.py ``` On 8 cores: expect 4-6x speedup. The CPU work parallelizes well; I/O is fast enough not to dominate. ## Step 6: Try alternatives, compare (10 min) Compare these versions: 1. Naive sync. 2. ThreadPoolExecutor (I/O only): some speedup but GIL limits. 3. ProcessPoolExecutor (CPU + I/O): big speedup. 4. asyncio + ProcessPoolExecutor (above): same as #3 for this workload. 5. ProcessPoolExecutor with smaller batches: per-record submission instead of per-file. Per-record is slower due to IPC