Algorithmic Problem Solving Questions

# Parameters
CHUNK_SIZE = bytes_fitting_in_memory  # read size per pass
NUM_PARTITIONS = k  # e.g., ~ M / avg_key_count

def read_chunks(file):
    while not eof:
        yield read_next_bytes(CHUNK_SIZE)

def local_aggregate(chunk_bytes):
    counts = {}
    for key, val in parse_kv_stream(chunk_bytes):
        counts[key] = counts.get(key, 0) + 1
    return counts  # fits in memory per chunk

# Phase 1: detect heavy hitters & partitioned spill
heavy_threshold = T  # e.g., 0.01 * total_records or empirical
heavy_estimates = defaultdict(int)
temp_partitions = [open('part_{}.tmp'.format(i),'ab') for i in range(NUM_PARTITIONS)]
for chunk in read_chunks(input_file):
    local = local_aggregate(chunk)
    # update heavy estimates (approx): maintain top-k in-memory
    for k,v in local.items():
        heavy_estimates[k] += v
    # if key is heavy, write to dedicated file; else hash-partition
    for k,v in local.items():
        if heavy_estimates[k] > heavy_threshold:
            write_to_file('heavy_{}.tmp'.format(k_hash(k)), k, v)
        else:
            pid = hash(k) % NUM_PARTITIONS
            write_to_file(temp_partitions[pid], k, v)
# close temp files

# Phase 2: reduce partitions (each fits in memory or process stream)
final_counts = {}
for pid in range(NUM_PARTITIONS):
    agg = {}
    for k,v in read_kv_from_file('part_{}.tmp'.format(pid)):
        agg[k] = agg.get(k,0) + v
    # write per-partition sorted counts
    write_sorted_counts('counts_{}.txt'.format(pid), agg)

# Process heavy hitters separately (incrementally)
for heavy_file in heavy_files:
    count = 0
    for k,v in read_kv_from_file(heavy_file):
        count += v
    write_heavy_count(k, count)

# Phase 3: external merge sorted partition outputs to produce globally sorted counts
# If each counts_*.txt is sorted by key, perform k-way merge; or collect all (key,count) and external sort by count if needed.

from pyspark.sql import functions as F, Window

# 1) determine date range needed (e.g., last 30 days)
dates_to_read = get_date_list(start_date, end_date)  # list of 'yyyy-MM-dd' strings

# 2) read only those partitions (partition pruning)
df = spark.read.parquet("/data/events/").filter(F.col("date").isin(dates_to_read))

# 3) map-side aggregation: compact to one row per user+date
user_date = (df
    .groupBy("date", "user_id")
    .agg(F.sum("amount").alias("daily_amount"))  # associative -> map-side combine used
)

# 4) repartition by user to colocate each user's rows, then compute ordered rolling sum
user_part = user_date.repartition("user_id")  # one shuffle but after heavy reduction

w = Window.partitionBy("user_id").orderBy(F.col("date").cast("timestamp")).rowsBetween(-N+1, 0)
result = user_part.withColumn("rolling_N", F.sum("daily_amount").over(w))

import heapq
from collections import defaultdict
from typing import Iterable, List, Tuple

def top_k_frequent(events: Iterable[str], k: int) -> List[Tuple[str,int]]:
    """
    Streaming-friendly: iterate events once to build counts,
    then maintain a min-heap of size k to get top-k frequent keys.
    Returns list of (key, count) sorted by descending count.
    """
    if k <= 0:
        return []

    counts = defaultdict(int)
    # Single pass over stream
    for e in events:
        counts[e] += 1

    # Min-heap of (count, key)
    heap = []
    for key, cnt in counts.items():
        if len(heap) < k:
            heapq.heappush(heap, (cnt, key))
        else:
            # if current count larger than smallest in heap, replace
            if cnt > heap[0][0]:
                heapq.heapreplace(heap, (cnt, key))

    # Convert to descending order
    result = sorted(((key, cnt) for cnt, key in heap), key=lambda x: -x[1])
    return result

# Example:
# events = ["a","b","a","c","b","a","d"]
# top_k_frequent(events, 2) -> [("a",3), ("b",2)]