Databricks-Spark-Assoc Developing DataFrame/DataSet API Applications — Timed 10 Questions

Question 1

You are processing a DataFrame containing nested JSON structures using Spark SQL functions. Which TWO of the following approaches allow you to flatten or extract fields from a struct column named 'user_info'? (Choose two)

10 minute time limit — choose an answer to begin.

Accepted Answer

df.select(col('user_info.name')). Extracting nested fields is a frequent requirement in data engineering pipelines when dealing with semi-structured sources. Using dot notation or the col function directly accesses struct fields, while explode unwraps arrays. Mastering these transformations ensures robust parsing of complex data types in production Databricks environments.

Answer

df.select(explode(col('user_info')))

Answer

df.select(json_tuple(col('user_info'), 'name'))

Answer

df.select(flatten(col('user_info')))

Question 2

You are joining two large DataFrames, 'sales' and 'products', on a common key 'product_id'. To ensure the join operation executes efficiently without causing driver memory issues, what is the best practice approach in Spark?

Accepted Answer

sales.join(broadcast(products), 'product_id'). Broadcast joins eliminate the shuffle phase by sending a copy of the smaller table to all worker nodes. This optimization prevents massive data movement across the network when joining a large fact table with a small dimension table, significantly boosting execution speed.

Answer

sales.join(products, 'product_id', 'cartesian')

Answer

sales.join(products, 'product_id').collect()

Answer

sales.hint('MERGE').join(products, 'product_id')

Question 3

A data engineer needs to aggregate sales data by store region and compute the running total of sales ordered chronologically within each region using window functions. Which DataFrame API pattern correctly accomplishes this without causing an out-of-memory error on the driver?

Accepted Answer

df.withColumn("running_total", sum("amount").over(Window.partitionBy("region").orderBy("date").rowsBetween(Window.unboundedPreceding, Window.currentRow))). Window specifications must define both a partitioning expression to isolate logical groups and an ordering clause to sequence rows deterministically for cumulative calculations. Applying these partitions properly distributes the memory footprint across executors, preventing driver memory exhaustion. This pattern is fundamental in Spark for handling time-series analytics and running totals efficiently in distributed architectures.

Answer

df.groupBy("region").agg(sum("amount").over(Window.partitionBy("region").orderBy("date")))

Answer

df.orderBy("date").agg(sum("amount").over(Window.partitionBy("region")))

Answer

df.select("*", sum("amount").over(Window.orderBy("region", "date")))

Question 4

An engineering team needs to process a large DataFrame containing user transaction logs. They must filter rows where the amount exceeds 500, and then extract the top 10 highest amounts without using global sorting. Which transformation sequence achieves this efficiently in PySpark?

Accepted Answer

df.filter(col("amount") > 500).orderBy(col("amount").desc()).limit(10). Using DataFrame.filter followed by DataFrame.take or similar localized operations avoids a costly global shuffle across the entire cluster. Global sorting via sort() followed by limit() forces a full exchange of data across partitions, which severely degrades performance on large datasets. Local sorting combined with aggregation or reduction limits network I/O and keeps execution stages optimized for distributed workloads in Apache Spark.

Answer

df.filter(col("amount") > 500).sort(col("amount").desc()).limit(10)

Answer

df.limit(10).filter(col("amount") > 500)

Answer

df.selectExpr("top_k(amount, 10)").filter(col("amount") > 500)

Question 5

A data engineer writes a PySpark application that reads a JSON file containing nested structures and arrays. They need to flatten the arrays into separate rows while preserving other field values. Which TWO DataFrame transformations should they use together in their pipeline? (Choose two)

Accepted Answer

pyspark.sql.functions.explode. Exploding nested array columns transforms each element in an array into a separate row while duplicating the other column values accordingly. When dealing with nested structures containing multiple arrays that need aligned flattening, joining or combining explode with structurally aware transformations ensures data integrity. Structuring pipelines with these methods is critical for transforming semi-structured JSON payloads into tabular formats suitable for downstream analytics.

Answer

pyspark.sql.functions.flatten

Answer

pyspark.sql.functions.coalesce

Answer

pyspark.sql.functions.to_json

Question 6

You have a PySpark DataFrame named df containing employee records with columns department, name, and salary. You need to calculate the running total of salaries for each department ordered by salary in ascending order, without reducing the number of output rows. Which DataFrame transformation should you apply?

Accepted Answer

df.select("*", sum("salary").over(Window.partitionBy("department").orderBy("salary"))). Window specifications in PySpark allow computing cumulative metrics like running totals without collapsing rows, unlike groupBy aggregations. By partitioning by department and ordering by salary, the Window function accurately computes the cumulative sum row by row while preserving the original granularity of the DataFrame.

Answer

df.groupBy("department").agg(sum("salary").over(Window.orderBy("salary")))

Answer

df.rollup("department", "salary").sum()

Answer

df.cube("department").agg(sum("salary"))

Question 7

A developer is building a PySpark job that reads a Parquet file into a DataFrame `events` and needs to add a monotonically increasing, unique ID column to each row for downstream deduplication. The DataFrame has no existing numeric column that guarantees uniqueness, and the job runs on a cluster with multiple executors. Which single DataFrame API call should they use to attach such an ID column without triggering a full shuffle of the existing data?

Accepted Answer

events.withColumn("row_id", monotonically_increasing_id()). monotonically_increasing_id() is the correct choice because it is a Catalyst-native expression that assigns unique IDs per row by combining the partition ID with an internal counter, requiring no shuffle. The window-based row_number() and RDD zipWithIndex both force data movement, while uuid() produces non-numeric, non-ordered values and does not satisfy the monotonic requirement.

Answer

events.withColumn("row_id", uuid())

Answer

events.withColumn("row_id", zipWithIndex())

Answer

events.withColumn("row_id", row_number().over(Window.orderBy(lit(1))))

Question 8

A data engineer has a PySpark DataFrame `df` with columns `order_id`, `customer_id`, and `order_ts` (a TimestampType column). The engineer must return only the rows where `order_ts` falls on or after January 1, 2023, and must keep the result as a DataFrame without triggering collection to the driver. Which transformation should be used?

Accepted Answer

df.filter(df.order_ts >= lit('2023-01-01').cast('timestamp')). A lazy DataFrame transformation such as `filter` with a correctly typed timestamp literal keeps processing distributed and lets Catalyst optimize the predicate. Building the comparison against a cast timestamp avoids type-mismatch errors, and no driver-side collection occurs. The other choices either misuse the API, drop to the slower RDD layer, or collect all rows, none of which satisfy the requirement.

Answer

df.orderBy('order_ts').collect() then filter in Python

Answer

df.select(df.order_ts.where('order_ts >= 2023-01-01'))

Answer

df.rdd.filter(lambda r: r.order_ts >= '2023-01-01').toDF()

Question 9

A developer is working with a DataFrame `df` that has a `status` string column containing values like 'active', 'inactive', and 'pending'. They need to replace every occurrence of 'inactive' with 'disabled' in that column while keeping all other values unchanged. Which operation should they use?

Accepted Answer

df.withColumn("status", regexp_replace("status", "inactive", "disabled")). Replacing a specific string value in place is a column transformation, and `regexp_replace` inside `withColumn` does it while preserving all rows and other columns. Filtering drops rows, aliasing does nothing, and `na.replace` targets nulls rather than general value substitution, so it is a poor fit for this scenario.

Answer

df.na.replace("inactive", "disabled", "status")

Answer

df.filter(col("status") != "inactive")

Answer

df.selectExpr("status AS status")

Question 10

A data engineer is building a PySpark application that reads a stream of JSON events from a Kafka topic into a DataFrame. The events have a nested struct field `device` containing `os` and `version`. The engineer needs to add a flattened column `os_version` that concatenates `device.os` and `device.version` with a hyphen, while keeping all original columns. Which code snippet correctly accomplishes this?

Accepted Answer

df = df.withColumn('os_version', concat(col('device.os'), lit('-'), col('device.version'))). Using withColumn with concat and col references for nested fields is the correct way to add a new column while preserving the original DataFrame. The other options misuse string literals or invalid indexing syntax. Only the option that uses col('device.os') and col('device.version') with concat will produce the desired concatenated column.

Answer

df = df.select('*', concat('device.os', lit('-'), 'device.version').alias('os_version'))

Answer

df = df.withColumn('os_version', concat(col('device')['os'], lit('-'), col('device')['version']))

Answer

df = df.withColumn('os_version', concat_ws('-', 'device.os', 'device.version'))