Data Transformation and Processing Logic Questions

Direct answer

Join skew means a small number of key values account for a disproportionate share of the rows, so whichever worker gets assigned that key becomes a straggler while everyone else finishes early. The standard toolkit, in order of how often each is reached for, is: let the engine's adaptive query execution handle it automatically if available, broadcast the smaller side to avoid a shuffle entirely, salt the skewed key to spread it across multiple partitions, or pre-aggregate the hot keys before the join.

Structured elaboration

1. Broadcast join: if one side is genuinely small enough to fit in memory on every worker, send a full copy to each worker and join locally with zero shuffle. This eliminates skew entirely, but only works when "small" is actually true, verify the real size in the query plan rather than trusting a size estimate, since a plan that broadcasts a table that was filtered AFTER the broadcast decision was made can overwhelm the driver.
2. Salting: append a random integer in [0, N) to the skewed key on the large side, replicate the corresponding row(s) on the small side N times (once per salt value), and join on the composite (key, salt); this spreads one hot key's rows across N partitions instead of one, at the cost of replicating the small side's matching rows N-fold.
3. Pre-aggregation of hot keys: if the join is followed by an aggregation anyway, partially aggregate the large side by key BEFORE the join, shrinking exactly the rows that were causing the skew in the first place.
4. Adaptive query execution (Spark 3.0+ and similar): modern engines can detect skewed partitions at runtime and automatically split them, which is worth mentioning first in an interview as the "check this before hand-rolling salting" step.
5. Column pruning and predicate pushdown: for a wide table (many columns) causing OOM independent of skew, select only the columns actually needed before the join and push filters down as early as possible, this is a complementary fix, not an alternative to the skew techniques above.

Worked example

A fact table with one hot key (1,000 rows) and four other keys (10 rows each), verified by execution:

pre-salt distribution:  key=1 -> 1000 rows, keys 2-5 -> 10 rows each

Salting key 1 into 4 buckets (salt = ROW_NUMBER() OVER (PARTITION BY key) % 4, then joining on (key, salt) after replicating the small side's matching row 4 times):

post-salt distribution for key=1: salt 0 -> 250, salt 1 -> 250, salt 2 -> 250, salt 3 -> 250

The 1,000-row hot key is now spread across 4 partitions of 250 rows each instead of landing entirely on one worker, verified by execution.

Trade-offs and pitfalls

• Salting only helps if you pick N large enough to actually resolve the skew (too small and each salted partition is still large) but not so large that the small-side replication becomes its own memory problem.
• A broadcast join can backfire when the size estimate the optimizer used is stale or was computed before a filter that would have shrunk the table; always check what the engine actually broadcasts via EXPLAIN, not what you assume it should.
• Salting changes the join's output cardinality bookkeeping (you must remember to strip the salt column afterward, and any downstream GROUP BY needs to happen after that, not before, or you'll double-count).

Direct answer

Read the file in fixed-size chunks using a generator, apply the transformation to each chunk as it's read, and yield (or persist) results incrementally, so memory use stays bounded by the chunk size regardless of the file's total size; periodically persist enough state to disk that a crash can resume from the last completed chunk rather than the beginning.

Structured elaboration

1. Generator, not a function that returns a list: a generator yields one chunk's results at a time, so the caller (or a downstream consumer) can process and discard each chunk before the next one is even read, which is what keeps memory bounded.
2. Chunk size as the memory/overhead knob: too small a chunk size adds per-chunk overhead without much memory benefit; too large defeats the purpose; size it as a fraction of the machine's available memory, not an arbitrary constant.
3. Handle real-world CSV messiness inline: header rows need to be read once and not treated as data; malformed lines need to be logged and skipped without crashing the whole read; different encodings need to be detected or configured rather than assumed.
4. Resumability: persist a marker (byte offset, or chunk index) after each successfully processed chunk, so a restart can skip straight to the first unprocessed chunk instead of reprocessing everything.
5. Backpressure, if downstream consumption is slower than reading: a generator naturally provides this for free (it only produces the next chunk when asked), which is one of the underrated reasons to prefer this shape over eagerly reading everything into memory and then filtering.

Worked example

def process_large_csv(file_path, chunk_size, transform_fn):
    with open(file_path) as f:
        header = f.readline()
        chunk = []
        for line in f:
            try:
                chunk.append(parse_row(line, header))
            except ValueError:
                continue  # malformed line skipped, logged elsewhere
            if len(chunk) >= chunk_size:
                yield transform_fn(chunk)
                chunk = []
        if chunk:
            yield transform_fn(chunk)

The chunked-processing shape itself was verified end to end (generating 250 synthetic records to a temp file plus one deliberately malformed line): processing correctly completed all 250 valid records across multiple chunks of at most 100 records held in memory at any point, with the malformed line skipped rather than crashing the run, confirming both the memory bound and the fault tolerance the pattern is meant to provide.

Trade-offs and pitfalls

• A generator-based approach only bounds memory if the CALLER also consumes it incrementally; wrapping the whole thing in list(process_large_csv(...)) defeats the entire point by materializing everything anyway.
• Resuming from a byte-offset checkpoint assumes the source file doesn't change between the crash and the restart; if the file could be appended to or rewritten in the interim, the checkpoint needs to also validate that the file's earlier content hasn't shifted underneath it.
• Encoding detection is a real source of subtle corruption (a file that's actually Latin-1 but is read as UTF-8 will parse "successfully" while silently mangling non-ASCII characters); when the encoding isn't guaranteed, validate it rather than assuming UTF-8 by default.

Direct answer

The right way to handle a column's missing values depends first on WHY the data is missing, not just on how much is missing; understanding whether the missingness is completely random (MCAR), related to observed data (MAR), or related to the missing value itself (MNAR) should drive the choice between dropping, imputing, or explicitly flagging missingness as its own category.

Structured elaboration

1. MCAR (missing completely at random): the fact that a value is missing has nothing to do with any variable, observed or not. Dropping rows is relatively safe here since it doesn't systematically bias the remaining data, though it still costs sample size.
2. MAR (missing at random, conditional on observed data): missingness correlates with something you CAN observe (e.g. income is more often missing for self-employed respondents, and you know who's self-employed). Imputation conditioned on the observed correlate (model-based, or at least stratified by the correlate) is more defensible than a single global mean/median.
3. MNAR (missing not at random): the missingness itself depends on the unobserved value (e.g. very high earners are more likely to decline to state income). This is the hardest case, naive imputation can systematically bias results, and an explicit "missing" indicator/category, or a domain-informed model of the missingness mechanism itself, is often more honest than pretending you can fill it in accurately.
4. When to add an indicator instead of imputing a value at all: if the fact that a value is missing is itself informative (which is common under MAR/MNAR), an explicit missing-indicator feature can carry real signal that a filled-in value would erase.
5. Communicating the choice: document which strategy was used per column and why, since a stakeholder consuming a dashboard or a model built on imputed data needs to know how much of what they're looking at is real versus filled-in, especially for columns with high missingness rates.

Worked example

A churn dataset with missing values in income, last_login, and plan_type, where the missingness patterns differ:

• plan_type missing only for very old accounts predating a schema migration: this looks close to MCAR (the migration is unrelated to any account's actual behavior), dropping or imputing with the overall mode is comparatively low-risk.
• last_login missing specifically for accounts that never logged in after signup: this is closer to MNAR-adjacent (the missingness IS the signal, "never logged in"), so an explicit "never logged in" category, not any numeric imputation, is the appropriate handling.
• income missing more often among certain segments who chose not to disclose it: closer to MAR/MNAR, a global mean/median fill would understate the real variation and should be flagged to stakeholders as a known limitation, or handled with a segment-conditioned estimate instead.

Trade-offs and pitfalls

• Defaulting to "just impute with the mean/median" for every column regardless of the missingness mechanism is the most common mistake; it's the easy answer but not always the correct one.
• An explicit missing-indicator column can leak information in a modeling context if the indicator itself was constructed using information not actually available at prediction time; treat it with the same point-in-time discipline as any other feature.
• Communicating "this metric is based on N% imputed values" honestly changes how much a stakeholder should trust a headline number, saying so is uncomfortable but is exactly the trust-preserving move a senior analyst makes.

Direct answer

Use PySpark's explode (or explode_outer if empty/null arrays must be preserved rather than dropped) on the nested array column, and select nested-struct fields with dot notation, wrapping both in defensive null-checks so a record with a missing or empty array doesn't fail the job.

Structured elaboration

1. explode vs explode_outer: plain explode drops any row whose array is null or empty (an inner-join-like behavior); explode_outer keeps that row with a NULL in the exploded column instead, which matters whenever downstream logic needs to know the parent record existed even if it had nothing in the array.
2. Nested struct access: PySpark's dot notation (col("payload.user.id")) navigates a nested struct directly once the JSON has been read with an inferred or explicit schema; optional/missing nested fields resolve to NULL rather than raising, as long as the schema marks them nullable.
3. Schema safety for optional fields: explicitly marking nested fields nullable in the schema (rather than relying purely on inference) avoids a job failure when a batch happens not to contain a field that a previous batch did.

Worked example

from pyspark.sql.functions import explode_outer, col

result = (
    raw_df
    .withColumn("tag", explode_outer(col("meta.tags")))
    .select(
        col("id"),
        col("user.id").alias("user_id"),
        col("tag"),
    )
)

The underlying explode-semantics were verified via an equivalent DuckDB UNNEST computation against a payload with a 3-element tags array and a payload with an empty tags array (PySpark itself is not installed in this sandbox): the 3-element array correctly produces 3 output rows, and the empty array correctly produces one output row with a NULL tag (matching explode_outer's behavior) rather than dropping that record entirely, which is exactly the distinction this answer's code makes between explode and explode_outer.

Trade-offs and pitfalls

• Using plain explode where explode_outer was actually needed silently drops every record whose array happens to be empty; this is easy to miss in testing if your test data never includes an empty-array case.
• Exploding two independent array columns from the same record in one step produces their cross product, not a paired iteration; if two arrays are meant to be paired index-by-index, arrays_zip (then a single explode over the zipped result) is the correct tool, not two separate explodes.
• Selecting a deeply nested field that doesn't exist in a GIVEN batch's schema (rather than being merely null) raises an analysis error unless the read step's schema was defined with that field as optional up front; this is a common source of a job that "worked yesterday" breaking today when a producer's payload shape shifts.

Direct answer

Maintain a bounded-memory structure that only remembers "recently seen" ids, evicting anything older than a configured TTL (time-to-live), so memory never grows unbounded even on an infinite stream. Two implementation choices trade off differently: an exact structure (a dict/ordered-map of id to last-seen timestamp) gives zero false positives but costs memory proportional to the number of DISTINCT ids within the TTL window; a Bloom filter caps memory at a fixed size regardless of cardinality, at the cost of a tunable false-positive rate.

Structured elaboration

1. Exact, TTL-bounded (LRU-style, least-recently-used): keep an ordered map from id to last-seen time. On each new event, first evict everything older than the TTL from the front of the ordering, then check membership, then record the new id at the back.
2. Probabilistic (Bloom filter): for very high cardinality where even a TTL-bounded exact map is too large, trade exactness for a fixed memory footprint; a false positive means a genuinely-new event gets incorrectly treated as a duplicate and dropped, which is the failure mode to disclose to whoever consumes the output.
3. Correctness consequence of each: exact eviction can let a true duplicate through if it arrives just after its TTL window closed (a real trade-off, not a bug, since unbounded memory is not an option); the Bloom filter can drop a true new event as a false duplicate, silently under-counting.

Worked example

from collections import OrderedDict

class StreamDeduper:
    def __init__(self, ttl_seconds):
        self.ttl = ttl_seconds
        self.seen = OrderedDict()  # id -> timestamp, insertion order approximates eviction order

    def offer(self, event_id, ts):
        self._evict(ts)
        if event_id in self.seen:
            return False
        self.seen[event_id] = ts
        return True

    def _evict(self, now):
        while self.seen:
            oldest_id, oldest_ts = next(iter(self.seen.items()))
            if now - oldest_ts > self.ttl:
                self.seen.pop(oldest_id)
            else:
                break

Run against the sequence ('a',0), ('b',1), ('a',2), ('a',150), ('b',151) with a 100-second TTL, this returns [True, True, False, True, True] (verified): the second 'a' at t=2 is correctly flagged as a duplicate of the one at t=0, but 'a' at t=150 is treated as new, because by then the original t=0 entry has aged out past the 100-second TTL and been evicted.

Trade-offs and pitfalls

• The TTL choice directly trades correctness against memory: too short and you miss real duplicates that arrive just late; too long and memory grows with the event rate times the window.
• A GC/cleanup pass for TTL-based state at real scale needs to run without blocking the offer/check path and without a race that lets a record get evicted and re-admitted as "new" mid-cleanup; a common pattern is a background sweep that only removes entries strictly older than the TTL, using the same clock the offer path uses.
• For a Bloom filter, false positives compound over time unless you periodically rebuild/rotate the filter (e.g. time-bucketed filters), or the effective false-positive rate creeps upward as more items are inserted than the filter was sized for.