Working with Large-Scale Datasets Questions

For a single-machine exploratory run on a 200GB Parquet dataset, both Dask DataFrame and Polars are strong options but trade off usability, performance, and memory strategies.

Comparison (pros / cons)

• Dask DataFrame
  • Pros: Familiar pandas-like API; lazy, out-of-core execution via partitions; integrates with existing PyData stack; can spill to disk and use distributed scheduler if needed.
  • Cons: Higher overhead per task, more Python-level scheduling cost; some pandas ops not implemented; memory overhead per partition.
• Polars
  • Pros: Rust-backed, very low memory overhead and high single-threaded throughput; efficient Arrow memory layout; good for columnar operations and fast CSV/Parquet IO; lazy query optimizer for pushdown and projection.
  • Cons: API differs from pandas in places (but becoming idiomatic); fewer ecosystem integrations; single-process by default (but multithreaded).

API differences

• Dask mirrors pandas: df.mean() → returns Dask object; must call .compute().
• Polars has eager and lazy APIs: pl.read_parquet (eager) or pl.scan_parquet (lazy) and .collect() for execution; expressions are used for transformations.

Memory strategies

• Dask: partition the dataset into many smaller pieces that fit memory; use worker memory limits and spilling to disk; incrementally aggregate (map-reduce) to limit peak memory.
• Polars: operate columnar and push down projections/filters to read less data; use lazy execution to build an optimized pipeline; relies on multithreading and zero-copy where possible — fewer allocations.

Examples — compute mean per numeric column

Dask:

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)

Polars (eager):

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)

Recommendation: If you want pandas-compatibility and out-of-core partitioning with easy incremental work, use Dask. For maximum single-machine performance and lower memory pressure on columnar ops, prefer Polars (use lazy scanning and projection to limit IO).

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.

Situation: You must join two customer tables (millions of rows) where names/IDs are inconsistent.

Approach (high level)

1. Clarify matching rules and acceptable error (precision vs recall). Get sample labeled pairs if possible to tune thresholds.

Preprocessing

• Normalize: lowercase, strip punctuation, unicode normalize, expand abbreviations (St. → Street), remove stopwords (“Inc”, “LLC” for orgs).
• Create canonical tokens: first/last, middle initials, join tokens.

Blocking / Candidate generation (scale)

• Deterministic blocks: exact match on normalized email domain, phone last 7 digits, or ZIP+first letter.
• Token/blocking: create blocking keys like first 4 chars of last name + birth year.
• Approximate blocking at scale:
  • Postgres: pg_trgm index for trigram similarity to quickly find close strings.
  • Spark: use MinHash LSH or BucketedRandomProjectionLSH on name n-gram TF vectors to get candidate pairs.
  • Elasticsearch: fuzzy prefix or n-gram index for candidate retrieval.
• Canopy clustering (cheap loose similarity to create overlapping blocks).

Candidate scoring

• Feature set: Jaro-Winkler, Levenshtein distance, trigram similarity, shared n-gram Jaccard, phonetic match (Metaphone/Soundex), exact matches on email/phone/address tokens, numerical ID presence.
• Combine features:
  • Rule-based weighted score (e.g., email exact = +0.6, phone match = +0.5, name Jaro-Winkler>0.9 = +0.4).
  • Or train a classifier (logistic regression / gradient boosted trees) using labeled pairs; use predicted probability as match score. Tools: Dedupe library (active learning + ML), Python recordlinkage, or Spark ML.

Choosing thresholds

• Use validation set to compute precision/recall and ROC/PR curves.
• Pick threshold based on business need: high precision for automated merges; lower threshold to surface manual review queue.
• Consider three-way decision: match / non-match / possible-match (manual review set sized by capacity).

Tools & implementation notes

• Postgres: pg_trgm + GIN index for similarity searches; fuzzystrmatch for soundex/metaphone.
• Spark: Scalability with MinHash LSH or TF-IDF + cosine, Spark ML for classifier training.
• Libraries: Dedupe (active learning, scalable), recordlinkage, Python jellyfish for phonetic/Jaro.
• Production: build candidate generation in Spark, store features in Parquet, train model offline, apply thresholding and write outputs; for interactive lookups use Elasticsearch.

Edge cases & best practices

• Monitor false matches, maintain explainable scores, log decisions for audits.
• Incremental matching: new records vs. existing index to avoid full cross-join.
• Evaluate on representative samples and tune blocking to balance recall and compute cost.

Approach: treat this as a trade-off decision + phased migration. First weigh costs and constraints, then identify queries to push down, then plan a safe, observable migration for pandas-only logic.

Decision criteria

• Network IO vs compute cost: pushing down reduces row transfer and Python memory/CPU but increases warehouse compute (often billed). Estimate bytes transferred vs warehouse query cost (use EXPLAIN/bytes-scanned metrics). Push down when transfer cost and local memory/latency dominate or warehouse cost is acceptable.
• Data freshness / latency: If pipeline needs latest rows in near-real-time, ensure warehouse latency and ingestion windows meet SLA. Local cached snapshots may be preferable for very low-latency ops.
• Skillset & maintainability: SQL-savvy teams and available warehouse features (window functions, approximate aggregates, UDFs) favor pushdown. If logic is complex Python, consider maintainability cost of rewriting vs keeping in pandas.
• Security & governance: centralizing in warehouse can simplify access controls, but may expose sensitive data if not handled properly.
• Observability & debugging: warehouses have different tooling; ensure tracing, explain plans, and cost visibility.

Query patterns that benefit from pushdown

• Wide filters/selectivity: heavy row pruning (WHERE predicates) to avoid transferring large tables.
• Aggregations/group-bys and joins: especially large joins or rollups that reduce data size early.
• Window functions, percentiles, time-series summarizations that warehouses optimize.
• Sorting/limit/top-k when few rows returned.
• Predicate pushdown and column-level projection.
• Pre-aggregation for ML feature stores (e.g., user-level aggregates).

Safe migration outline for pandas-specific logic

1. Profiling & prioritization: measure pipeline stages (memory, CPU, time, bytes). Pick high-memory, deterministic stages first.
2. Reimplement in SQL when possible: map pandas idioms to SQL (groupby->GROUP BY, rolling->WINDOW, merge->JOIN). Use warehouse-native functions for performance.
3. Hybrid approach: use SQL to prefilter/aggregate to reduce data, then run remaining pandas transforms on a small sample locally.
4. For custom Python transforms:
   • Option A: Move as UDFs/Stored Procedures if warehouse supports Python UDFs (evaluate performance and security).
   • Option B: Containerized UDFs / external function frameworks (e.g., Snowflake External Functions, BigQuery Remote UDFs) to keep Python logic but run near data.
   • Option C: Keep in pandas but run on a downsampled/pre-aggregated dataset fetched from warehouse.
5. Validation & testing:
   • Build parity tests: run SQL version and pandas version on same inputs and assert identical outputs (or acceptable numeric tolerance).
   • Use data diff tools and unit tests for edge cases (NaNs, dtype changes, timezone).
   • Incremental rollout: enable pushdown for a subset of data or traffic, compare metrics and costs.
6. Observability & cost control:
   • Add logging of bytes scanned, runtime, and cost per query; set alerts and query quotas.
   • Keep query plans and use ANALYZE/EXPLAIN to spot inefficiencies.
7. Documentation & knowledge transfer: document SQL translations, maintainability notes, and rollback plan.

Trade-offs and final guidance

• Favor pushdown for heavy data reduction operations. Keep complex, iterative Python-only logic where rewriting cost, correctness risk, or UDF overhead outweighs benefits.
• Always quantify cost/benefit (transfer bytes vs warehouse dollars) and instrument metrics so you can reverse decisions if cost or latency regressions appear.

Approach (brief):

• We need a sort-merge join. One CSV (A) is already sorted by join key. The other (B) is unsorted and too large to fit memory, so externally sort B into a sorted stream (create sorted runs that fit memory, then k-way merge).
• Then stream A and sorted-B in lockstep: advance the stream with smaller key; when keys equal, buffer the full group of identical-key rows from each side (careful about group size; spill to disk if the group exceeds memory), produce Cartesian product of group pairs and write results with a buffered CSV writer.

Python sketch (practical but simplified):

import csv, tempfile, heapq, os
from itertools import islice, product

def external_sort_csv(path, key_idx, max_rows_in_memory=100000, tmpdir=None):
    # Phase 1: read chunks, sort, write runs
    runs = []
    with open(path, newline='') as f:
        rdr = csv.reader(f)
        header = next(rdr)
        while True:
            chunk = list(islice(rdr, max_rows_in_memory))
            if not chunk: break
            chunk.sort(key=lambda r: r[key_idx])
            tf = tempfile.NamedTemporaryFile(delete=False, dir=tmpdir, mode='w', newline='')
            writer = csv.writer(tf)
            writer.writerow(header)
            writer.writerows(chunk)
            tf.close()
            runs.append(tf.name)
    # Phase 2: k-way merge generator (skip headers)
    def merged_iter():
        files = [open(r, newline='') for r in runs]
        readers = [csv.reader(f) for f in files]
        # skip headers
        for rd in readers: next(rd, None)
        heap = []
        for i, rd in enumerate(readers):
            try:
                row = next(rd)
                heap.append((row[key_idx], i, row))
            except StopIteration:
                pass
        heapq.heapify(heap)
        while heap:
            key, i, row = heapq.heappop(heap)
            yield row
            try:
                r = next(readers[i])
                heapq.heappush(heap, (r[key_idx], i, r))
            except StopIteration:
                pass
        for f in files: f.close()
        # cleanup run files
        for r in runs: os.remove(r)
    return header, merged_iter()

def sort_merge_join(path_sortedA, path_unsortedB, key_idxA, key_idxB, out_path, mem_rows=100000):
    # get B as sorted iterator
    headerB, sortedB_iter = external_sort_csv(path_unsortedB, key_idxB, max_rows_in_memory=mem_rows)
    fA = open(path_sortedA, newline='')
    rdrA = csv.reader(fA)
    headerA = next(rdrA)
    out = open(out_path, 'w', newline='')
    writer = csv.writer(out)
    writer.writerow(headerA + headerB)  # simple concat
    # iterators that yield rows
    itA = iter(rdrA)
    itB = iter(sortedB_iter)

    def read_group(it, key_idx, first_row=None):
        # collect group of same-key rows; returns key, list(rows)
        rows = []
        if first_row:
            k = first_row[key_idx]; rows.append(first_row)
        else:
            try:
                r = next(it)
            except StopIteration:
                return None, []
            k = r[key_idx]; rows.append(r)
        while True:
            try:
                peek = next(it)
            except StopIteration:
                return k, rows
            if peek[key_idx] == k:
                rows.append(peek)
            else:
                # we consumed one row that belongs to next group -> need to stash it
                # wrap it back by making a generator that yields peek then rest
                def chain(first, rest):
                    yield first
                    for x in rest: yield x
                new_it = chain(peek, it)
                return k, rows, new_it
        # unreachable

    # simpler streaming loop using one-row lookahead
    a_row = next(itA, None)
    b_row = next(itB, None)
    while a_row and b_row:
        a_key = a_row[key_idxA]; b_key = b_row[key_idxB]
        if a_key < b_key:
            a_row = next(itA, None)
        elif a_key > b_key:
            b_row = next(itB, None)
        else:
            # gather groups
            a_group = [a_row]
            b_group = [b_row]
            # drain A same-key
            while True:
                nxt = next(itA, None)
                if nxt and nxt[key_idxA] == a_key:
                    a_group.append(nxt)
                else:
                    a_row = nxt
                    break
            # drain B same-key
            while True:
                nxt = next(itB, None)
                if nxt and nxt[key_idxB] == b_key:
                    b_group.append(nxt)
                else:
                    b_row = nxt
                    break
            # produce cross-product; if groups huge, consider streaming partial products and/or spill
            for ra, rb in product(a_group, b_group):
                writer.writerow(ra + rb)
    fA.close(); out.close()

Key concepts and reasoning:

• External sort (sort runs + k-way merge) enables sorting B with limited memory.
• Sort-merge join streams sorted inputs, requiring only buffering of rows sharing the same key (group-by-key).
• Duplicate keys (many-to-many) require materializing groups; if groups exceed memory, spill groups to temp files and then stream Cartesian product in blocks.
• Efficient I/O: read/write in chunks, use csv module and buffered files, remove temp files promptly.

Complexity:

• I/O dominates: external sort cost ~ O(N log_M N) disk IO; merge join is linear passes over both files.
• Memory: controlled by max_rows_in_memory; additional memory for buffering equal-key groups.

Edge cases:

• Different key types/formatting: normalize keys before compare.
• Massive key groups: implement spill-to-disk for groups and compute join in block-wise streaming to avoid O(group_size^2) memory.
• Null/missing keys: decide join semantics (skip or treat as value).