Python for Data Analysis Questions

import pandas as pd
from collections import Counter

def compute_metrics(path, chunksize=100_000):
    """
    Reads CSV in chunks and computes:
    - unique user count (exact, using a set)
    - top 10 products by total sales (using Counter for aggregation)
    Assumes CSV has columns: user_id, product_id, sale_amount
    """
    user_set = set()
    product_sales = Counter()

    # Specify dtypes and usecols to reduce memory footprint
    for chunk in pd.read_csv(
        path,
        usecols=["user_id", "product_id", "sale_amount"],
        dtype={"user_id": str, "product_id": str, "sale_amount": float},
        chunksize=chunksize
    ):
        # Clean / drop malformed rows
        chunk = chunk.dropna(subset=["user_id", "product_id", "sale_amount"])

        # Update unique users
        user_set.update(chunk["user_id"].unique())

        # Aggregate sales per product in this chunk, then merge into global counter
        chunk_agg = chunk.groupby("product_id", sort=False)["sale_amount"].sum()
        for prod, amt in chunk_agg.items():
            product_sales[prod] += float(amt)

    unique_user_count = len(user_set)
    top_10_products = product_sales.most_common(10)

    return unique_user_count, top_10_products

# Example usage:
# users, top_products = compute_metrics("large_sales.csv", chunksize=200_000)
# print(users, top_products)

import time
from collections import OrderedDict, defaultdict, deque
import threading
import pickle

class CountMinSketch:
    def __init__(self, width=2000, depth=5, seed=1):
        # simple placeholder; production use multiple hash funcs
        self.width, self.depth = width, depth
        self.table = [[0]*width for _ in range(depth)]
    def _hashes(self, key):
        for i in range(self.depth):
            yield (hash((key,i)) % self.width)
    def add(self, key, delta=1):
        for i, idx in enumerate(self._hashes(key)):
            self.table[i][idx] += delta
    def estimate(self, key):
        return min(self.table[i][idx] for i, idx in enumerate(self._hashes(key)))

class StreamingAggregator:
    def __init__(self, max_hot=100_000, allowed_lateness=300):
        self.max_hot = max_hot
        self.allowed_lateness = allowed_lateness
        self.hot = OrderedDict()   # key: (user_id, hour_ts) -> count
        self.cms_per_hour = defaultdict(lambda: CountMinSketch())
        self.wal = open('wal.log','ab')  # append-only
        self.lock = threading.Lock()
    def _hour_bucket(self, event_time):
        return event_time - (event_time % 3600)
    def ingest(self, user_id, event_time, watermark_ts):
        hour = self._hour_bucket(event_time)
        if event_time + self.allowed_lateness < watermark_ts:
            # too late: route to late path (emit correction or drop)
            self._handle_late(user_id, hour)
            return
        key = (user_id, hour)
        with self.lock:
            if key in self.hot:
                self.hot.move_to_end(key)
                self.hot[key] += 1
            else:
                if len(self.hot) >= self.max_hot:
                    self._evict_one()
                self.hot[key] = 1
            # write minimal WAL record for recovery
            pickle.dump(('inc', user_id, hour, 1), self.wal)
            self.wal.flush()
    def _evict_one(self):
        evicted_key, count = self.hot.popitem(last=False)
        user_id, hour = evicted_key
        # merge into CMS for that hour
        self.cms_per_hour[hour].add(user_id, count)
        # optionally persist a snapshot of evicted exact counts
    def emit_hourly(self, emit_fn, current_watermark):
        """Emit aggregates for windows that are complete given watermark"""
        to_emit = []
        with self.lock:
            # collect all hours older than watermark - allowed_lateness
            cutoff = current_watermark - self.allowed_lateness
            hours = set(h for (_,h) in self.hot.keys()) | set(self.cms_per_hour.keys())
            for hour in hours:
                if hour + 3600 <= cutoff:  # window closed
                    # gather from hot exact counts
                    per_user = {}
                    for (uid,h), cnt in list(self.hot.items()):
                        if h == hour:
                            per_user[uid] = per_user.get(uid,0) + cnt
                            del self.hot[(uid,h)]
                    # combine with CMS estimates for other users when needed
                    cms = self.cms_per_hour.pop(hour, None)
                    # emit exact hot users
                    for uid,cnt in per_user.items():
                        emit_fn(uid, hour, cnt, exact=True)
                    # emit approximate top-K or aggregated cold counts could be done by estimating for demanded users
                    # here we could emit aggregate-only metrics or leave CMS for on-demand queries
                    to_emit.append(hour)
        return to_emit
    def checkpoint(self):
        with self.lock:
            # snapshot hot and cms
            with open('snapshot.pkl','wb') as f:
                pickle.dump((self.hot, self.cms_per_hour), f)

    def recover(self):
        # reload snapshot and replay WAL tail
        pass

    def _handle_late(self, user_id, hour):
        # policy: accept and update state if within a larger bounded window, or emit to late topic
        # For simplicity, write to late-log
        with open('late.log','ab') as f:
            pickle.dump(('late', user_id, hour), f)

import dask.dataframe as dd
# read parquet as partitioned Dask dataframe
df = dd.read_parquet("data/*.parquet")  # partitions by file
# select numeric columns then compute column means (returns pandas Series)
means = df.select_dtypes(include=['number']).mean().compute()
print(means)

import polars as pl
# lazy scan to avoid reading unnecessary columns immediately
lf = pl.scan_parquet("data/*.parquet")
# get numeric columns and compute means via expressions
numeric_cols = lf.schema.items()
# compute means for all numeric columns; collect executes query
means_df = lf.select([pl.col(pl_datatype).mean().alias(name)
                      for name, pl_datatype in lf.schema.items()
                      if pl_datatype in (pl.Int64, pl.Float64)]).collect()
print(means_df)

import pandas as pd
import json
import logging
import time
from pathlib import Path
from typing import Dict, List

logger = logging.getLogger("etl.loader")

class SchemaError(Exception):
    pass

def _read_file(path: Path, max_retries=2, backoff=1.0) -> pd.DataFrame:
    tries = 0
    while True:
        try:
            suffix = path.suffix.lower()
            if suffix in (".csv", ".tsv"):
                return pd.read_csv(path)
            if suffix in (".parquet",):
                return pd.read_parquet(path)
            if suffix in (".json",):
                return pd.read_json(path, lines=True)
            raise ValueError(f"Unsupported format: {suffix}")
        except (OSError, pd.errors.EmptyDataError) as e:
            if tries < max_retries:
                tries += 1
                logger.warning("Transient read error, retrying: %s (try %d)", e, tries)
                time.sleep(backoff * tries)
                continue
            logger.exception("Failed to read file after %d tries", tries+1)
            raise

def load_and_validate(path: str,
                      required_cols: Dict[str, str],  # col -> pandas dtype (e.g. "int64", "float")
                      coerce: bool = True) -> pd.DataFrame:
    path = Path(path)
    df = _read_file(path)

    # Check missing required columns
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        msg = f"Missing required columns: {missing}"
        logger.error(msg)
        raise SchemaError(msg)

    # Coerce types
    for col, dtype in required_cols.items():
        try:
            if str(df[col].dtype) != dtype:
                if coerce:
                    df[col] = df[col].astype(dtype, errors="raise")
                    logger.info("Coerced column %s to %s", col, dtype)
                else:
                    raise SchemaError(f"Column {col} has dtype {df[col].dtype}, expected {dtype}")
        except (ValueError, TypeError) as e:
            # Recoverable? Attempt safe conversion for numeric with NaNs
            if coerce and dtype.startswith(("int", "float")):
                logger.warning("Coercion warning for %s: %s. Using pd.to_numeric with coercion", col, e)
                df[col] = pd.to_numeric(df[col], errors="coerce")
            else:
                logger.exception("Fatal schema mismatch for column %s", col)
                raise SchemaError(f"Cannot coerce column {col} to {dtype}: {e}")

    # Additional lightweight validations
    # e.g., drop rows with critical nulls
    critical_nulls = df[required_cols.keys()].isnull().all(axis=1).sum()
    if critical_nulls:
        logger.warning("Found %d rows with all required columns null; dropping", critical_nulls)
        df = df.dropna(subset=list(required_cols.keys()), how="all")

    return df