Working with Large-Scale Datasets Questions

HyperLogLog (HLL) is a probabilistic sketch for cardinality estimation. It hashes each user_id to a uniform 64-bit value, splits the hash into an index (p bits → m=2^p registers) and a suffix used to count leading zeros. Each register stores the max observed "zero-run" length. The cardinality estimate is computed from the harmonic mean of registers, with bias corrections. Crucial properties: very small fixed memory per sketch, mergeable (register-wise maximum), and relative error ≈ 1.04 / sqrt(m).

Implementation approach (daily sketches + merge):

• For each event, feed user_id into an HLL update for that event’s date (streaming or batch).
• Persist one HLL sketch per date.
• To get DAU over any window (e.g., 365 days) merge the daily sketches (register-wise max) and extract cardinality.

Error bounds and memory:

• Choose p to trade error vs memory. Example: p=14 → m=16384, error ≈ 1.04/√16384 ≈ 0.81%. Memory: approx m * bits_per_register. Typical implementations pack 6-bit registers → ~12 KB per sketch. 365 days → ~4.4 MB total — tiny versus storing raw IDs.

Mergeability:

• Merge = elementwise max of registers; associative & commutative → safe to merge many daily sketches in any order (ideal for distributed processing).

Practical caveats:

• Small-cardinality bias: use exact counting when counts are small (many libs handle this).
• Hash quality matters; use 64-bit hash.
• If you need tight SLAs, pick p accordingly or combine with a Bloom filter / exact store for top users.

Example SQL / pseudocode

Postgres with hll extension:

-- daily aggregation into table daily_hll(day date, sketch hll)
INSERT INTO daily_hll
SELECT date_trunc('day', ts) as day,
       hll_add_agg(hll_hash_integer(user_id)) as sketch
FROM events
GROUP BY 1;

-- merge 365 sketches and get unique users
SELECT hll_cardinality(hll_union_agg(sketch)) as dau_365
FROM daily_hll
WHERE day BETWEEN '2024-01-01' AND '2024-12-31';

BigQuery using HLL functions:

-- create daily HLLs
CREATE TABLE project.dataset.daily_hll AS
SELECT DATE(timestamp) AS day,
       HLL_COUNT.INIT(user_id) AS sketch
FROM `project.dataset.events`
GROUP BY day;

-- merge and extract
SELECT HLL_COUNT.EXTRACT(HLL_COUNT.MERGE_AGG(sketch)) AS dau_365
FROM project.dataset.daily_hll
WHERE day BETWEEN '2024-01-01' AND '2024-12-31';

Alternative (BigQuery APPROX_COUNT_DISTINCT):

SELECT APPROX_COUNT_DISTINCT(user_id) AS dau_365
FROM `project.dataset.events`
WHERE DATE(timestamp) BETWEEN ...

(approx single-pass, not reusable per-day sketches)

Why this helps:

• Building per-day sketches is cheap and incremental; merging 365 small sketches is fast and memory efficient; you avoid expensive per-day distinct scans over raw tables. Choose p based on target relative error and storage budget (p between 12–16 common).

Approach (high-level)

• Use Spark built-in aggregations (mean, stddev) with groupBy + agg for scale; push as much work to executors (avoid collecting to driver).
• For median (per-group quantile) use approximate algorithms: Spark's percentile_approx (SQL) or approxQuantile per column; for many groups use a distributed sketch (t-digest or DataSketches) computed per-partition and merged per-group.
• Handle skew with salting, map-side combining, and repartitionByRange/hash to balance hot keys.

PySpark examples

1. Mean & Stddev (scales, uses Spark aggregations)

from pyspark.sql import functions as F

df = spark.read.parquet("/data/events")  # columns: key, value

agg_stats = (
    df.groupBy("key")
      .agg(
          F.count("*").alias("n"),
          F.mean("value").alias("mean"),
          F.stddev_pop("value").alias("stddev")
      )
)

2. Median using percentile_approx (fast, approximate)

median_df = (
    df.groupBy("key")
      .agg(F.expr("percentile_approx(value, 0.5, 1000) as median"))  # 1000 interpolation steps
)

3. Scalable exact-ish quantiles using t-digest sketches (recommended for many groups)

• Compute per-partition t-digest for each group, merge across partitions with reduceByKey, then extract quantiles on executors (no driver memory blowup).

# requires tdigest package on workers
from tdigest import TDigest
from pyspark.sql import Row

rdd = df.select("key", "value").rdd.map(lambda r: (r['key'], float(r['value'])))

def sketches_in_partition(iter_rows):
    from tdigest import TDigest
    d = {}
    for key, v in iter_rows:
        t = d.get(key)
        if t is None:
            t = TDigest()
            d[key] = t
        t.update(v)
    for key, t in d.items():
        yield (key, t)

# map partitions to produce partial digests, then merge by key
partial = rdd.mapPartitions(sketches_in_partition)
merged = partial.reduceByKey(lambda a, b: a.merge(b) or a)  # TDigest.merge returns merged digest
# compute desired quantiles per key
result = merged.mapValues(lambda td: {"median": td.percentile(50), "p10": td.percentile(10)})
result_df = result.map(lambda kv: Row(key=kv[0], **kv[1])).toDF()

Key techniques and reasoning

• Use built-in aggregations for mean/stddev because they are optimized (treeAggregate/combine) and avoid driver memory issues.
• For median, exact sorting per group is expensive. Use approximate quantile algorithms: percentile_approx (Spark SQL), approxQuantile (column-level), or streaming sketches (t-digest, DataSketches) for per-group quantiles. Sketches are mergeable and keep memory bounded.
• Skew handling:
  • Salting: add random prefix to hot keys to split work then un-salt and merge partial aggregates.
  • Pre-aggregate with map-side combine (use combiners in RDD or aggregators in Dataset).
  • Repartition by key to increase parallelism and avoid single executor overload.
  • Monitor and tune shuffle partitions (spark.sql.shuffle.partitions).

Complexity & trade-offs

• mean/stddev with groupBy: time ~ O(n) with shuffle; memory per executor proportional to distinct keys in partition.
• percentile_approx: O(n) but constant additional memory; accuracy controlled by parameter (higher -> more memory).
• t-digest sketch: O(n) time, small bounded memory per key; highly accurate for medians/percentiles and mergeable.
• Trade-offs: exact median requires full sort per group (expensive). Approximate methods trade slight accuracy for huge performance and memory gains.

Edge cases

• Extremely high-cardinality keys: consider filtering, sampling, or hierarchical aggregation (aggregate to coarser buckets).
• Very hot single key: use salting and dedicated resources.
• Nulls and NaNs: filter or handle explicitly before aggregation.
• Library availability: ensure tdigest or DataSketches are shipped to executors (--py-files or spark.jars).

This plan balances accuracy, performance, and memory constraints for computing grouped mean/std/median at scale in PySpark.

Approach summary:

• Assumptions: stream of integers too large to store; we can store O(k) counters. We want the top-k frequent items approximately. I’ll describe a brute-force exact method (store full counts) then implement the Misra–Gries (Frequent) algorithm which is deterministic, memory-efficient, and returns candidate heavy-hitters. I’ll include complexity and error bounds, plus practical notes.

Brute-force exact method:

• Keep a hashmap count[item] incrementing for each stream element. After the stream, return top-k by sorting counts. Exact but uses O(U) memory where U = number of distinct items — impractical for large streams.

Approximate memory-efficient algorithm: Misra–Gries (k-counter variant)

• Guarantees: any item with true frequency > N/(k+1) will appear among the ≤k candidates. For any returned candidate, the reported count is at most its true count and at least true_count − N/(k+1).
• Space: O(k). Single pass, deterministic.

Python implementation (Misra–Gries):

from collections import defaultdict

class MisraGries:
    def __init__(self, k):
        assert k >= 1
        self.k = k
        self.counters = {}  # item -> count

    def process(self, x):
        # Process one item from stream
        if x in self.counters:
            self.counters[x] += 1
            return
        if len(self.counters) < self.k:
            self.counters[x] = 1
            return
        # Decrement all counters by 1; remove zeros
        to_delete = []
        for key in list(self.counters.keys()):
            self.counters[key] -= 1
            if self.counters[key] == 0:
                to_delete.append(key)
        for key in to_delete:
            del self.counters[key]

    def process_stream(self, iterable):
        for x in iterable:
            self.process(x)

    def get_candidates(self):
        # Returns current candidates with their (under-)estimated counts
        return dict(self.counters)

Usage:

• Instantiate MisraGries(k). Feed stream via process or process_stream. The counters map gives candidate items with reduced counts. For more accurate ranking, a second pass over the stream can compute exact counts for only the candidate set (if re-reading is allowed) and then pick true top-k among them.

Time & Space complexity:

• Time: O(N * k) worst-case for naive decrement step each new distinct when counters full; with k small this is acceptable. With implementation that stores counters in dict and decrements k entries, amortized O(N * k). Space: O(k).

Error bounds and guarantees:

• If N is total number of items seen, for any item i with true frequency f_i:
  • If f_i > N/(k+1), item i must be among the returned candidates.
  • The estimated count c_i from the algorithm satisfies: max(0, f_i − N/(k+1)) ≤ c_i ≤ f_i.
• To obtain exact counts for the candidates, do a second pass counting only those items (space O(k), time O(N)).

Alternative (Count–Min Sketch) notes:

• CMS uses t hash functions and w counters per row. Space O(t*w). Gives one-sided overestimates: estimated >= true count, with error ≤ εN with probability 1−δ when w = ceil(e/ε) and t = ceil(ln(1/δ)). Useful when you need approximate frequencies with probabilistic guarantees and sublinear space, but misra-gries is simpler for top-k candidate extraction.

Edge cases:

• Very large k relative to distinct items (works fine).
• If heavy hitters have frequency near threshold, may need second-pass verification.
• Stream with adversarial patterns can make operation costlier but memory bound remains O(k).

This implementation is simple, practical for data-science pipelines, and pairs well with a second-pass verification to report exact top-k.

Requirements & constraints:

• Detect near-duplicates among 100M short text records (names/addresses/descriptions) tolerant to typos, token reorderings, and small edits.
• High recall on duplicates, reasonable precision, scalable (single big machine or distributed), incremental updates.

High-level approach:

1. Blocking (coarse candidate generation)
   • Create multiple cheap blocking keys to ensure recall: normalized tokens, sorted tokens, phonetic codes (Soundex/Metaphone), n-gram shingles, and domain-specific rules.
   • Use multi-pass blocking (each pass uses a different key) to catch different error modes.
2. Similarity representation
   • For token-order invariance and small edits: represent records as sets of character n-grams (e.g., 3-grams) or token sets.
   • Optionally compute vectorized string embeddings (FastText/transformer-based) for semantic similarity; combine with syntactic measures.
3. Dimensionality reduction & LSH/MinHash
   • For set-based signatures compute MinHash signatures (e.g., 128 hashes) to approximate Jaccard; index via LSH (banding) to produce buckets of likely near-duplicates.
   • For dense embeddings use cosine LSH (SimHash or Random Projection LSH) or approximate nearest neighbor (ANN) index (Faiss, Annoy, HNSW).
4. Candidate refinement & clustering
   • For each candidate pair from blocks/LSH/ANN compute exact similarity metrics: normalized edit distance, Jaccard, cosine, weighted combination. Apply threshold and transitive clustering (connected components / single-linkage) with careful merging heuristics to avoid chaining errors.
5. Scalability & implementation
   • Single big machine: stream records, compute shingles and MinHash in batches (memory-fit), store LSH buckets on-disk or in-memory key-value store; use Faiss/HNSW for embedding ANN.
   • Distributed: map-reduce style pipeline: Map phase computes normalized forms, shingles, MinHash bands or embeddings; shuffle by band keys to reducers that do pairwise comparisons per bucket. Use Spark/Dataproc for distributed MinHash/LSH or Faiss GPU clusters for embeddings.
6. Practical considerations
   • Tune MinHash size and LSH banding for precision/recall trade-off; prune overly large buckets; maintain stop-word lists and frequency-based blocking to avoid heavy buckets.
   • Incremental updates: append-only LSH buckets and ANN indexes with periodic reindexing or use dynamic HNSW structures.
   • Evaluation: sample-based precision/recall, pairwise labeling, and end-to-end deduplication F1.
     Example tooling: Python with datasketch (MinHash, LSH), Faiss/HNSWlib for ANN, Spark for distributed processing, RocksDB/S3 for storage.

This design balances cheap blocking (high recall) with LSH/ANN to limit pairwise checks, and exact similarity + clustering for high-precision deduplication at 100M scale.

Approach (brief): Build a Bloom filter from the small dimension table's key column, broadcast it to workers, use it to pre-filter incoming streaming events so only likely-matching rows go through the expensive join. This reduces shuffle and memory pressure at the cost of occasional false positives (no false negatives).

Spark example (PySpark Structured Streaming):

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf
from pyspark.sql.types import BooleanType

spark = SparkSession.builder.getOrCreate()

# Load small dimension table (assume static / small enough)
dim = spark.read.parquet("/path/dim_keys")  # schema: (key, attr1,...)

# Build Bloom filter: set expected items and false positive probability
expected_items = 5_000_000
fpp = 0.01
bf = dim.stat.bloomFilter("key", expected_items, fpp)

# Broadcast Bloom filter to executors
bf_broadcast = spark.sparkContext.broadcast(bf)

# UDF to test membership
@udf(BooleanType())
def bf_contains(k):
    if k is None:
        return False
    return bf_broadcast.value.mightContain(k)

# Streaming events
events = spark.readStream.format("kafka").option(...).load() \
    .selectExpr("CAST(value AS STRING) as json") \
    .selectExpr("from_json(json, 'key STRING, payload STRUCT<...>') as data") \
    .select(col("data.key").alias("key"), col("data.payload"))

# Pre-filter using Bloom filter
filtered = events.filter(bf_contains(col("key")))

# Then join with dim (broadcast join if dim small)
joined = filtered.join(dim, on="key", how="left")

# Write out
query = joined.writeStream.format("parquet").option(...).start()

Why this helps:

• Pre-filtering removes rows that are guaranteed not to match, lowering the number of records entering the join and thus reducing network shuffle and downstream memory usage.
• Best when one side (dim) is much smaller and you expect many streaming events with keys not present in dim.

False positives & tuning:

• Bloom filters can return false positives but never false negatives. That means some non-matching events will still be forwarded and cost join work, but no legitimate matches are dropped.
• Tune expected_items and fpp: choose expected_items close to real distinct key count; choose fpp (e.g., 0.001–0.01) balancing memory vs extra work. Memory ≈ -(n * ln(p)) / (ln(2)^2) bits; lower p → more bits.
• If expected_items underestimated, filter becomes saturated → higher fpp. If overestimated, wasted memory.

Effect on shuffle and memory:

• Reduced shuffle volume because fewer rows participate in the join; especially effective when pre-filter rejects a large fraction of events.
• Bloom filter itself consumes memory (stored and broadcast). For very large expected_items, Bloom filter size may be non-trivial and costly to broadcast.
• For streaming, update strategy: for dynamic dims you must rebuild and rebroadcast the filter periodically or maintain a scalable structure (e.g., counting Bloom or partitioned filters).
• If dim is small enough, prefer broadcasting the full table and doing a broadcast hash join — Bloom filter is most useful when dim is moderately small but you still have huge event volume and need cheap negative filtering before shuffle.

Trade-offs / best practices:

• Measure selectivity: if only a small fraction of events are filtered, overhead of checking and broadcasting may not pay off.
• Keep Bloom filter size fitting executor memory and network transfer limits.
• Combine with partition pruning and local deduplication to maximize benefits.
• Monitor real false-positive rate and rebuild filter as dim cardinality changes.