PySpark Interview Questions with Examples

import re

from pyspark.sql import SparkSession, Window, functions as F, types as T

spark = (SparkSession.builder.master("local[4]").config("spark.ui.showConsoleProgress", "false")
         .config("spark.driver.memory", "2g").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
tracker = spark.sparkContext.statusTracker()
PATH = "household_power_consumption.txt"                       # UCI household power: ';'-separated, '?' = missing


def jobs(group, action):
    spark.sparkContext.setJobGroup(group, group)
    out = action()
    return out, len(tracker.getJobIdsForGroup(group))


print("P1 schema inference and a placeholder for missing values")
inferred, n_inferred = jobs("infer", lambda: spark.read.csv(PATH, sep=";", header=True, inferSchema=True))
print(f"   inferSchema=True: Global_active_power is {dict(inferred.dtypes)['Global_active_power']} (jobs to read the schema: {n_inferred})")
with_null, _ = jobs("infer-null", lambda: spark.read.csv(PATH, sep=";", header=True, inferSchema=True, nullValue="?"))
print(f"   with nullValue='?': Global_active_power is {dict(with_null.dtypes)['Global_active_power']}")

print("\nP2 an explicit schema skips the inference pass")
schema = T.StructType([T.StructField("Date", T.StringType()), T.StructField("Time", T.StringType())]
                      + [T.StructField(c, T.DoubleType()) for c in ["Global_active_power", "Global_reactive_power", "Voltage",
                                                                    "Global_intensity", "Sub_metering_1", "Sub_metering_2", "Sub_metering_3"]])
power, n_explicit = jobs("explicit", lambda: spark.read.csv(PATH, sep=";", header=True, schema=schema, nullValue="?"))
print(f"   jobs to read with an explicit schema: {n_explicit}")

print("\nP3 nulls")
total, missing = power.count(), power.filter(F.col("Global_active_power").isNull()).count()
avg_all = power.agg(F.avg("Global_active_power")).first()[0]
avg_zero = power.fillna(0, ["Global_active_power"]).agg(F.avg("Global_active_power")).first()[0]
print(f"   {total:,} rows, {missing:,} with no reading | avg ignoring nulls {avg_all:.4f} kW | avg after fillna(0) {avg_zero:.4f} kW")

print("\nP4 parsing dates: the pattern must match the data")
sample = power.select("Date").distinct().orderBy("Date")
print(f"   example raw values: {[r.Date for r in sample.limit(3).collect()]}")
try:
    power.select(F.to_date("Date", "dd/MM/yyyy").alias("d")).filter(F.col("d").isNull()).count()
except Exception as error:                                     # Spark 3 refuses rather than silently returning nulls
    name = type(error).__name__
    reason = next(line for line in str(error).splitlines() if "could not be parsed" in line or "Fail to parse" in line)
    message = re.sub(r"'[^']*'", "'<first bad value>'", reason.split("You can set")[0].strip())   # which row fails first varies
    print(f"   'dd/MM/yyyy' -> {name}: {message}")
loose = power.select(F.to_date("Date", "d/M/yyyy").alias("d")).filter(F.col("d").isNull()).count()
print(f"   'd/M/yyyy' -> rows that fail to parse: {loose:,}")

daily = (power.withColumn("day", F.to_date("Date", "d/M/yyyy")).groupBy("day")
         .agg(F.avg("Global_active_power").alias("kw")).filter(F.col("kw").isNotNull()))
w = Window.orderBy("day").rowsBetween(-6, 0)
rolled = daily.withColumn("kw_7d", F.avg("kw").over(w)).orderBy(F.desc("kw")).first()
print(f"\nP5 window: highest daily average {rolled.kw:.3f} kW on {rolled.day}, 7-day rolling average then {rolled.kw_7d:.3f} kW")

print("\nP6 a Python UDF vs a built-in function")
watts_udf = F.udf(lambda kw: None if kw is None else kw * 1000.0, T.DoubleType())
by_udf = power.select(watts_udf("Global_active_power").alias("w"))
by_builtin = power.select((F.col("Global_active_power") * 1000).alias("w"))
same = by_udf.agg(F.sum("w")).first()[0] == by_builtin.agg(F.sum("w")).first()[0]
print(f"   same total: {same} | UDF plan runs Python (BatchEvalPython): {'BatchEvalPython' in by_udf._jdf.queryExecution().executedPlan().toString()}"
      f" | built-in plan does: {'BatchEvalPython' in by_builtin._jdf.queryExecution().executedPlan().toString()}")

print("\nP7 take() vs collect()")
print(f"   take(5) returned {len(power.take(5))} rows; collect() would pull all {total:,} rows into the driver")
spark.stop()

P1 schema inference and a placeholder for missing values
   inferSchema=True: Global_active_power is string (jobs to read the schema: 2)
   with nullValue='?': Global_active_power is double

P2 an explicit schema skips the inference pass
   jobs to read with an explicit schema: 0

P3 nulls
   2,075,259 rows, 25,979 with no reading | avg ignoring nulls 1.0916 kW | avg after fillna(0) 1.0779 kW

P4 parsing dates: the pattern must match the data
   example raw values: ['1/1/2007', '1/1/2008', '1/1/2009']
   'dd/MM/yyyy' -> Py4JJavaError: Fail to parse '<first bad value>' in the new parser.
   'd/M/yyyy' -> rows that fail to parse: 0

P5 window: highest daily average 3.315 kW on 2006-12-23, 7-day rolling average then 1.817 kW

P6 a Python UDF vs a built-in function
   same total: True | UDF plan runs Python (BatchEvalPython): True | built-in plan does: False

P7 take() vs collect()
   take(5) returned 5 rows; collect() would pull all 2,075,259 rows into the driver