Python Data Manipulation with Pandas & PySpark Questions

Techniques for cleaning, transforming, and analyzing data in Python using Pandas and PySpark. Covers working with DataFrames, data wrangling, missing-value handling, filtering, aggregations, joins, grouping, and typical patterns for data preparation and exploratory analysis, including both in-memory Pandas workflows and distributed PySpark processing.

HardTechnical

60 practiced

Implement a PySpark solution to compute sessionization for user events: create session_id per user where a new session starts if gap between events > 30 minutes. Provide code using window functions and explain stateful vs stateless approaches and how each behaves in streaming vs batch.

Sample Answer

Approach: For batch/sessionization we can use window functions to compute the time difference between consecutive events per user, mark session boundaries where gap > 30 minutes, and create cumulative session IDs. For streaming, similar logic applies but you must consider stateful processing (maintain last event timestamp per user across micro-batches) or use structured streaming session windows.

python

from pyspark.sql import functions as F
from pyspark.sql import Window

# df schema: user_id (string), event_time (timestamp), ... 
# Batch example
w = Window.partitionBy("user_id").orderBy("event_time")

df2 = (df
  .withColumn("prev_ts", F.lag("event_time").over(w))
  .withColumn("gap_minutes", 
              (F.unix_timestamp("event_time") - F.unix_timestamp("prev_ts"))/60)
  .withColumn("is_new_session", F.when(F.col("prev_ts").isNull() | (F.col("gap_minutes") > 30), 1).otherwise(0))
  # cumulative sum to assign session number per user
  .withColumn("session_n", F.sum("is_new_session").over(Window.partitionBy("user_id").orderBy("event_time")
                                                      .rowsBetween(Window.unboundedPreceding, 0)))
  .withColumn("session_id", F.concat(F.col("user_id"), F.lit("_"), F.col("session_n")))
)

Key points:- This is stateless: computes sessions from full dataset (batch) using window operations.- Complexity: sorting per user dominates; time O(N log N) for partition sorts; space depends on shuffle.- Edge cases: out-of-order timestamps (sort fixes in batch), duplicate timestamps, timezone normalization, late data.

Streaming vs batch and stateful vs stateless:- Batch/stateless (above): deterministic, easier — needs full history or whole partition to compute gaps.- Streaming/stateless (structured streaming session windows): use built-in session window aggregations (e.g., groupBy(session_window("event_time", "30 minutes")), but those create sessions per key automatically within the micro-batch and use event-time/watermarks to close sessions.- Streaming/stateful: maintain per-user last event timestamp in mapState or updateStateByKey equivalent to continue sessions across micro-batches. This is necessary when events arrive incrementally and you must decide whether an incoming event continues an existing session (compare to stored last_ts). Add watermarks to evict old state and avoid unbounded state growth.

Trade-offs:- Stateful streaming gives low-latency, correct incremental sessionization but requires careful state management, checkpoints, and TTL.- Batch is simpler and robust for historical recomputation but not suitable for low-latency needs.

Implementation tips:- Use event-time and watermarks in streaming to handle late data.- Persist session_id deterministically (based on first event timestamp) to allow idempotent updates.- Test with out-of-order/late events and set appropriate watermark thresholds.

MediumTechnical

58 practiced

A PySpark job writes many tiny Parquet files partitioned by date/hour and downstream jobs suffer from small-file overhead. Describe three strategies to reduce the small-file problem and show code or Spark configs that perform coalescing/compaction during writes.

Sample Answer

Approach summary: three practical strategies — 1) control parallelism at write (repartition/coalesce by partition keys), 2) limit records-per-file and tune shuffle/configs, 3) periodic compaction (merge small files after ingest, ideally using a transactional layer like Delta). Below are explanations, trade-offs, and PySpark examples.

1) Repartition / coalesce per partition at write- Ensure each target partition (date/hour) is written with a small, controlled number of output files by repartitioning on the partition keys then coalescing.- Trade-off: can introduce a shuffle; use when downstream read cost outweighs shuffle cost.

Example:

python

# ensure one file per date-hour (or small N files)
df = spark.read.format("parquet").load("s3://bucket/staging/")
# repartition by partition columns to route rows to correct task,
# then coalesce to reduce per-partition files (coalesce reduces tasks without shuffle)
df_repart = df.repartition("date", "hour").coalesce(1)  # coalesce(1) -> 1 file per spark task (per executer partition)
(df_repart
 .write
 .mode("append")
 .partitionBy("date", "hour")
 .option("compression", "snappy")
 .parquet("s3://bucket/processed/"))

Tip: use coalesce(N) instead of 1 for higher throughput (e.g., N = number of target files per partition).

2) Limit records per file and tune shuffle settings- Use write option maxRecordsPerFile to prevent many tiny files; tune spark.sql.shuffle.partitions to avoid too many small tasks.- Trade-off: maxRecordsPerFile doesn't guarantee equal split by partition keys but reduces tiny files.

Example:

python

spark.conf.set("spark.sql.shuffle.partitions", "200")  # reduce/increase based on data size
(df
 .repartition("date", "hour")   # route to partition keys
 .write
 .mode("append")
 .partitionBy("date", "hour")
 .option("maxRecordsPerFile", 2000000)  # ~2M rows per file (adjust)
 .option("compression", "snappy")
 .parquet("s3://bucket/processed/"))

3) Periodic compaction / merge job (recommended for streaming or frequent small writes)- Run scheduled compaction that reads affected partitions and rewrites them into larger files. Use Delta Lake OPTIMIZE if available — it handles transactionality and statistics.- Trade-off: extra periodic job and read+write cost; can be scheduled during off-peak.

Delta example (best when using Delta):

python

# Using Delta: one-line compaction
from delta.tables import DeltaTable
delta_tbl = DeltaTable.forPath(spark, "/mnt/delta/table")
# optimize specific partition (requires Delta >= supported version)
spark.sql("OPTIMIZE delta.`/mnt/delta/table` WHERE date = '2025-12-01' AND hour = '03'")

Parquet compaction example (non-Delta):

python

# compact specific partition into e.g. 4 files
p = "s3://bucket/processed/date=2025-12-01/hour=03"
df_part = spark.read.parquet(p)
(df_part.coalesce(4)            # choose desired number of files
        .write
        .mode("overwrite")
        .option("compression","snappy")
        .parquet(p))

Additional operational tips:- Prefer partition granularity aligned to access patterns (e.g., date vs date/hour) to avoid too many partitions.- Monitor file counts and average file size; target ~128MB file size for HDFS/S3-read efficiency.- Use transactional formats (Delta, Iceberg) to compact safely and avoid read inconsistencies.- Automate compaction for partitions with many small files using size/file-count heuristics.

These strategies combined (write-time tuning + scheduled compaction + proper partitioning) reduce small-file overhead while balancing shuffle and compute costs.

MediumTechnical

64 practiced

Given a Pandas DataFrame 'df' with a JSON column 'properties' (strings contain JSON objects), write Python (Pandas) code to expand the JSON into separate columns and handle missing keys by filling with NaN. Show code using pandas.json_normalize or apply + pd.Series.

Sample Answer

Approach: parse the JSON strings in the 'properties' column, expand each JSON object into separate columns and join back to the DataFrame. Use pandas.json_normalize for speed and consistency; provide an alternative using apply + pd.Series. Missing keys become NaN automatically.

python

import json
import pandas as pd

# example df
df = pd.DataFrame({
    "id": [1, 2, 3],
    "properties": [
        '{"a": 10, "b": "x"}',
        '{"b": "y", "c": 3.14}',
        None
    ]
})

# 1) Using json_normalize (recommended for nested JSON)
# parse JSON strings to dicts (handle None / invalid JSON)
props = df["properties"].apply(lambda s: json.loads(s) if isinstance(s, str) else None)
expanded = pd.json_normalize(props)  # missing keys => NaN
result = pd.concat([df.drop(columns=["properties"]), expanded], axis=1)

python

# 2) Alternative: apply + pd.Series
# safer for simple flat JSON; slightly slower at scale
props_series = df["properties"].apply(lambda s: pd.Series(json.loads(s)) if isinstance(s, str) else pd.Series())
result2 = pd.concat([df.drop(columns=["properties"]), props_series], axis=1)

Key points:- json_normalize handles nested structures (use record_path/meta for lists).- Missing keys produce NaN by default.- Validate/try-except json.loads if data may be malformed.Complexity: O(n * avg_json_size) CPU, memory O(n * avg_num_keys). For very large datasets use chunked processing or Spark.

EasyTechnical

52 practiced

You're given a CSV with 10M rows and you need to perform lightweight cleaning (drop unused columns, parse dates) before handing to analysts. In Pandas, what are three ways to reduce memory usage while reading this CSV? Provide code examples for dtype specification and for reading in chunks.

Sample Answer

Three simple, effective ways to reduce memory when reading a 10M-row CSV in Pandas:

1) Specify dtypes (and use category for low-cardinality strings) so Pandas doesn't upcast to object/float64 unnecessarily.2) Limit columns and parse dates while reading (usecols + parse_dates + infer_datetime_format) to avoid post-processing and extra memory.3) Read in chunks (chunksize) and process/serialize each chunk (e.g., clean and write to parquet) rather than loading all rows at once.

Examples:

- dtype specification (important for large numeric and categorical columns)

python

import pandas as pd

dtypes = {
    "id": "int32",
    "user_id": "int32",
    "amount": "float32",
    "country": "category",   # low-cardinality string -> category saves a lot
    "status": "category"
}

df = pd.read_csv("big.csv", usecols=["id","user_id","amount","country","status","created_at"],
                 dtype=dtypes,
                 parse_dates=["created_at"],
                 infer_datetime_format=True)

- reading in chunks and doing lightweight cleaning per chunk

python

import pandas as pd

chunks = pd.read_csv("big.csv", usecols=["id","user_id","amount","country","created_at"],
                     dtype={"id":"int32","user_id":"int32","amount":"float32","country":"category"},
                     parse_dates=["created_at"],
                     chunksize=500_000,  # tune to available memory
                     infer_datetime_format=True)

# Example: clean each chunk and append to parquet
for i, chunk in enumerate(chunks):
    # drop rows with missing amount, simple cleaning
    chunk = chunk.dropna(subset=["amount"])
    # optional: further conversions
    chunk.to_parquet(f"cleaned_part_{i}.parquet", index=False)

Notes / best practices:- Use usecols to drop unused columns during read (saves IO + memory).- Categories reduce memory if cardinality << rows; don't use category for high-cardinality unique IDs.- Write cleaned data to an efficient format (parquet) for analysts.

HardTechnical

97 practiced

Describe a robust approach in PySpark for schema evolution when appending nightly data to an existing Parquet dataset: how to detect new fields, align types, and safely append without corrupting downstream consumers. Include practical steps, code snippets, or tooling suggestions.

Sample Answer

Start by framing requirements: detect new/missing fields, reconcile type differences (safe promotions), preserve column ordering/nullable metadata, and atomically append so downstream consumers (queries, tables) never see partial or incompatible data.

Approach (practical steps)1. Discover schemas- Read existing Parquet schema (example: latest partition or table schema from Hive/Glue/Delta)- Read incoming nightly schema (infer or use provided schema file)

2. Compute reconciliation plan- Identify new columns (in incoming but not existing) → add as nullable to existing schema when reading older partitions- Identify removed columns → keep them nullable in new data to avoid breaking consumers- Type diffs → apply safe promotion rules (int -> long, float -> double, string win if incompatible); flag unsafe downgrades for manual review

3. Implement safe alignment in PySpark- Load incoming DataFrame; cast/add columns to match target schema; ensure column order consistent- Write to a staging location partitioned like production; validate row counts & schema- Atomically move staging partition into production (S3 atomic rename or use table-level transaction via Delta/Iceberg/Hudi)

Code (PySpark)

python

from pyspark.sql import functions as F
from pyspark.sql.types import *

# example: load target schema from table
target_df = spark.table("db.prod_table").limit(0)
target_schema = {f.name: f.dataType for f in target_df.schema}

incoming = spark.read.parquet("/data/nightly/part-*.parquet")
incoming_schema = {f.name: f.dataType for f in incoming.schema}

# add missing target cols to incoming
for col, dtype in target_schema.items():
    if col not in incoming_schema:
        incoming = incoming.withColumn(col, F.lit(None).cast(dtype))

# add new incoming cols to target (so older partitions get nulls) -- applied when reading historical data or via schema evolution
# align types with safe promotion
def safe_cast(df, col, dst_type):
    try:
        return df.withColumn(col, F.col(col).cast(dst_type))
    except:
        # fallback to string to avoid data loss and flag for review
        return df.withColumn(col, F.col(col).cast(StringType()))

for col, dst in target_schema.items():
    incoming = safe_cast(incoming, col, dst)

# reorder to target order
incoming = incoming.select(*[c for c in target_schema.keys()])

# write to staging and validate
staging = "/data/staging/prod_table/date=2025-12-06"
incoming.write.mode("overwrite").parquet(staging)

# validation step (row counts, schema equality)
# then atomically swap: on S3 use manifest-based swap or use Delta for transactional merge

Tooling recommendations- Use Delta Lake / Apache Iceberg / Hudi for ACID, schema evolution, time travel, and transactional atomic commits (preferred).- Maintain canonical table schema in Glue/Hive metastore or a schema registry (JSON/Avro) and CI checks for nightly schema diffs.- Implement automated alerts for unsafe type changes; require manual approval.

Governance & testing- Enforce tests: unit tests for schema reconciliation, integration test on staging, data quality checks (null rates, value ranges).- Keep a migration log + change requests for schema changes tracked in Git/PR.

Why this works- Explicit reconciliation prevents silent data corruption.- Staging + validation + atomic swap prevents downstream consumers from seeing partial/heterogeneous data.- Using transactional table formats (Delta/Iceberg/Hudi) simplifies merges and maintains compatibility.

Unlock Full Question Bank

Get access to hundreds of Python Data Manipulation with Pandas & PySpark interview questions and detailed answers.

Join thousands of developers preparing for their dream job.