Engineering: Python, SQL, PySpark & REST APIs

import time
import functools
import logging
from typing import Generator, Dict, Any, Callable

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("DE_Pipeline")

def retry_with_backoff(retries: int = 3, backoff_in_seconds: float = 1.0):
    """Decorator for resilient I/O operations with exponential backoff and jitter."""
    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            delay = backoff_in_seconds
            for attempt in range(1, retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logger.warning(f"Attempt {attempt}/{retries} failed for {func.__name__}: {e}")
                    if attempt == retries:
                        raise
                    time.sleep(delay)
                    delay *= 2.0
        return wrapper
    return decorator

def stream_large_dataset(file_path: str, chunk_size: int = 10000) -> Generator[list[str], None, None]:
    """Lazy streaming generator keeping memory strictly bounded at O(chunk_size)."""
    batch = []
    with open(file_path, "r", encoding="utf-8") as f:
        # Skip header
        header = f.readline().strip().split(",")
        for line in f:
            batch.append(line.strip())
            if len(batch) >= chunk_size:
                yield batch
                batch = []
        if batch:
            yield batch

# Production usage demonstration
if __name__ == "__main__":
    logger.info("Initialized memory-bounded ingestion generator.")

-- Compute Daily Portfolio NAV, High-Water Mark and Maximum Drawdown
WITH daily_nav AS (
    SELECT 
        portfolio_id,
        trade_date,
        SUM(shares_held * closing_price) AS market_value,
        SUM(cash_balance) AS cash_value,
        SUM(shares_held * closing_price + cash_balance) AS total_nav
    FROM investment_positions
    WHERE trade_date >= '2026-01-01'
    GROUP BY portfolio_id, trade_date
),
ranked_metrics AS (
    SELECT 
        portfolio_id,
        trade_date,
        total_nav,
        -- Running Peak-to-Date (High-Water Mark)
        MAX(total_nav) OVER (
            PARTITION BY portfolio_id 
            ORDER BY trade_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS high_water_mark,
        -- Prior day NAV for daily return calculation
        LAG(total_nav, 1) OVER (
            PARTITION BY portfolio_id 
            ORDER BY trade_date
        ) AS prior_day_nav
    FROM daily_nav
)
SELECT 
    portfolio_id,
    trade_date,
    total_nav,
    high_water_mark,
    ROUND(((total_nav - prior_day_nav) / NULLIF(prior_day_nav, 0)) * 100, 4) AS daily_return_pct,
    -- Drawdown percentage from the all-time peak
    ROUND(((total_nav - high_water_mark) / NULLIF(high_water_mark, 0)) * 100, 2) AS drawdown_pct
FROM ranked_metrics
ORDER BY portfolio_id, trade_date;

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("ProductionEngineeringTuning") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

def join_skewed_streams(transactions_df, merchants_df, salt_buckets: int = 16):
    """
    Eliminates straggler tasks when merchant_id has massive key skew 
    (e.g., Amazon, Apple accounting for 40% of all volume).
    """
    # 1. Salt the heavy skewed transaction stream
    salted_transactions = transactions_df.withColumn(
        "salt", F.floor(F.rand() * salt_buckets)
    ).withColumn(
        "salted_key", F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt"))
    )

    # 2. Replicate merchant dimension rows across all salt buckets
    replicated_merchants = merchants_df.withColumn(
        "salt", F.explode(F.array([F.lit(i) for i in range(salt_buckets)]))
    ).withColumn(
        "salted_key", F.concat(F.col("merchant_id"), F.lit("_"), F.col("salt"))
    )

    # 3. Join on the salted composite key
    result_df = salted_transactions.join(
        replicated_merchants,
        on="salted_key",
        how="inner"
    ).drop("salt", "salted_key")

    return result_df

import time
import requests
import hashlib
from typing import Dict, Any, Generator

class ResilientAPIExtractor:
    def __init__(self, base_url: str, api_key: str, requests_per_minute: int = 60):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}", "Accept": "application/json"}
        self.min_interval = 60.0 / requests_per_minute
        self.last_request_time = 0.0

    def _rate_limit(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_request_time = time.time()

    def fetch_paginated_records(self, endpoint: str) -> Generator[Dict[str, Any], None, None]:
        url = f"{self.base_url}/{endpoint}"
        cursor = None

        while url:
            self._rate_limit()
            params = {"cursor": cursor} if cursor else {}
            
            try:
                response = requests.get(url, headers=self.headers, params=params, timeout=10)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 10))
                    time.sleep(retry_after)
                    continue

                response.raise_for_status()
                data = response.json()

                for record in data.get("items", []):
                    # Compute deterministic idempotency hash
                    record_hash = hashlib.sha256(str(record).encode()).hexdigest()
                    record["_ingest_hash"] = record_hash
                    yield record

                cursor = data.get("next_cursor")
                if not cursor:
                    break

            except requests.RequestException as err:
                # Log and route to DLQ
                print(f"API Extraction Error: {err}")
                break