Optimising a Slow Data Processing Pipeline

# Current slow approach
import csv
import psycopg2

conn = psycopg2.connect(...)
cur = conn.cursor()

with open("data.csv") as f:
    reader = csv.DictReader(f)
    for row in reader:                    # 10M iterations
        validated = validate(row)
        transformed = transform(validated)
        cur.execute(                       # 10M individual INSERT statements!
            "INSERT INTO records VALUES (%s, %s, %s)",
            (transformed["id"], transformed["name"], transformed["value"])
        )
    conn.commit()

# Optimised approach — chunked bulk insert + multiprocessing

import csv
import psycopg2
from psycopg2.extras import execute_values
from concurrent.futures import ProcessPoolExecutor
import itertools

def process_chunk(rows: list[dict]) -> list[tuple]:
    """Validate and transform a chunk of rows — runs in subprocess."""
    result = []
    for row in rows:
        try:
            validated = validate(row)
            transformed = transform(validated)
            result.append((transformed["id"], transformed["name"], transformed["value"]))
        except ValueError:
            pass  # Skip invalid rows
    return result

def chunked(iterable, size):
    it = iter(iterable)
    while chunk := list(itertools.islice(it, size)):
        yield chunk

CHUNK_SIZE = 10_000

with open("data.csv") as f:
    reader = csv.DictReader(f)
    with ProcessPoolExecutor(max_workers=8) as pool:
        conn = psycopg2.connect(...)
        cur = conn.cursor()

        for processed_chunk in pool.map(process_chunk, chunked(reader, CHUNK_SIZE)):
            execute_values(
                cur,
                "INSERT INTO records (id, name, value) VALUES %s ON CONFLICT DO NOTHING",
                processed_chunk,
                page_size=1000
            )
        conn.commit()

# Results:
# Before: 4 hours (10M individual inserts, single thread)
# After:  ~18 minutes (bulk insert + 8 parallel processes)