Databricks-Spark-Assoc Developing DataFrame/DataSet API Applications Practice Test 3 — 15 Questions

Question 1

A data engineer has a PySpark DataFrame `orders` with a string column `order_ts` formatted as `yyyy-MM-dd HH:mm:ss`. They need a new column `order_date` containing only the date portion as a `date` type, so downstream code can filter by day. Which approach is correct?

Choose an answer to begin — your selection is scored in the full session.

Accepted Answer

`orders.withColumn('order_date', to_date('order_ts', 'yyyy-MM-dd HH:mm:ss'))`. Converting a formatted timestamp string to a `date` requires an explicit parse with a pattern. `to_date(col, pattern)` parses the string and returns a `date` column, which is the correct type for day-level filters and joins. String manipulation or `from_unixtime` yields strings, and `date_trunc` yields a timestamp, so none of them match the requirement as cleanly or as safely as `to_date`.

Answer

`orders.withColumn('order_date', from_unixtime(unix_timestamp('order_ts', 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd'))`

Answer

`orders.withColumn('order_date', date_trunc('day', 'order_ts'))`

Answer

`orders.withColumn('order_date', substring('order_ts', 1, 10))`

Question 2

A data engineer has a PySpark DataFrame `orders` with a string column `order_ts` formatted as `yyyy-MM-dd HH:mm:ss`. They need a new column `order_date` containing only the date portion as a `date` type, so downstream code can filter by day. Which approach is correct?

Choose an answer to begin — your selection is scored in the full session.

Accepted Answer

events.withColumn("prev_ts", lag("event_ts").over(Window.partitionBy("user_id").orderBy("event_ts"))). A window function with `partitionBy("user_id").orderBy("event_ts")` defines a frame scoped to each user in chronological order, and `lag` reads the value one row back inside that frame. That yields the same user's prior event timestamp on every row, returning null only for each user's earliest event, which matches the requested output exactly.

Answer

events.withColumn("prev_ts", lead("event_ts").over(Window.orderBy("event_ts")))

Answer

events.groupBy("user_id").agg(max("event_ts").alias("prev_ts"))

Answer

events.orderBy("event_ts").withColumn("prev_ts", first("event_ts"))

Question 3

A data engineer has a PySpark DataFrame `orders` with a string column `order_ts` formatted as `yyyy-MM-dd HH:mm:ss`. They need a new column `order_date` containing only the date portion as a `date` type, so downstream code can filter by day. Which approach is correct?

Choose an answer to begin — your selection is scored in the full session.

Accepted Answer

readings.select("sensor_id", (col("celsius") * 9/5 + 32).alias("fahrenheit")). The requirement is a column-wise transformation that preserves sensor_id and adds fahrenheit, evaluated across executors. select() with a Column expression and alias is the canonical DataFrame API approach: it is lazy, Catalyst-optimized, and does not move data to the driver. Converting to the RDD API is unnecessary, dropping celsius changes the schema, and collect() pulls everything to the driver, which the scenario forbids.

Answer

readings.collect().map(lambda r: (r.sensor_id, r.celsius * 9/5 + 32))

Answer

readings.rdd.map(lambda r: (r.sensor_id, r.celsius * 9/5 + 32)).toDF(["sensor_id", "fahrenheit"])

Answer

readings.withColumn("fahrenheit", col("celsius") * 9/5 + 32).drop("celsius")

Question 4

A developer must join a 4 TB `transactions` DataFrame against a 900 MB `merchants` DataFrame on `merchant_id`. The cluster has 40 executors each with 16 GB of memory, and the job currently shuffles the large side. They want to avoid the shuffle entirely. Which change should they make?

Accepted Answer

Call `broadcast(merchants)` when joining so the small DataFrame is shipped to every executor.. An explicit broadcast hint forces Spark to collect the small DataFrame and distribute it as a read-only hash map to every executor, so the large side is scanned in place with no exchange. Because the small table fits comfortably in executor memory, this converts a shuffle-heavy sort-merge join into a map-side lookup and removes the network bottleneck.

Answer

Increase `spark.sql.autoBroadcastJoinThreshold` only, without calling `broadcast`.

Answer

Repartition both DataFrames by `merchant_id` before the join.

Answer

Set `spark.sql.shuffle.partitions` to 40 so the shuffle matches the executor count.

Question 5

A developer has a DataFrame `raw` and needs to permanently persist it as Parquet partitioned by `region`, overwriting any existing data at that path, without registering it in the metastore. Which call achieves this?

Accepted Answer

raw.write.mode("overwrite").partitionBy("region").parquet("/data/out"). The DataFrameWriter path with `mode("overwrite")`, `partitionBy("region")`, and `.parquet` writes typed Parquet files into per-region directories and replaces prior contents in one step. Because it targets a filesystem path rather than a table name, no metastore entry is created, satisfying the "without registering it in the metastore" constraint.

Answer

raw.rdd.saveAsTextFile("/data/out")

Answer

raw.write.format("parquet").saveAsTable("out")

Answer

raw.createOrReplaceTempView("out").write.parquet("/data/out")

Question 6

A developer needs to join `orders` (large) with `customers` (small, a few thousand rows) on `customer_id`. They want to broadcast the small table and confirm the broadcast actually took effect. Which combination of actions is correct?

Accepted Answer

Call `orders.join(broadcast(customers), "customer_id")` and verify by inspecting the physical plan for `BroadcastHashJoin`. Broadcasting a small table avoids shuffling the large one, and the `broadcast()` hint is the explicit way to request it. Verification matters because hints can be ignored when the table exceeds the threshold or when the join type is unsupported. Inspecting the physical plan for `BroadcastHashJoin` is the standard confirmation, making the combination of hint plus plan inspection the correct answer.

Answer

Increase `spark.sql.autoBroadcastJoinThreshold` to a very large value and rely on the optimizer without checking the plan

Answer

Call `customers.repartition(1)` before the join so the small table is in one partition

Answer

Use `orders.join(customers.hint("shuffle_hash"), "customer_id")` and check the plan for `ShuffledHashJoin`

Question 7

A developer needs to add a column `full_name` to DataFrame `people` by concatenating `first_name` and `last_name` with a single space, and must handle rows where `last_name` is null by producing just the first name. Which expression is correct?

Accepted Answer

people.withColumn("full_name", concat_ws(" ", col("first_name"), col("last_name"))). `concat_ws` inserts the separator between non-null arguments and treats nulls as absent, so a null last name produces a clean first-name-only value with no dangling space. That matches the requirement for both the populated and the null case in one expression, without needing any explicit conditional branching.

Answer

people.withColumn("full_name", col("first_name") + lit(" ") + col("last_name"))

Answer

people.withColumn("full_name", when(col("last_name").isNull(), col("first_name")).otherwise(col("first_name")))

Answer

people.withColumn("full_name", concat(col("first_name"), lit(" "), col("last_name")))

Question 8

A data engineer has a DataFrame `events` with columns `user_id` and `ts`. They need to add a column `prev_ts` that holds the previous event timestamp for each user, ordered by `ts` ascending, without collapsing rows. Which operation accomplishes this?

Accepted Answer

Window.partitionBy('user_id').orderBy('ts') with lag('ts', 1) via withColumn. Using a window partitioned by `user_id` and ordered by `ts` with the `lag` function returns the previous event's timestamp for each row while retaining all rows. The other choices either aggregate away detail, rely on partition-local sorting that cannot guarantee correct ordering, or drop rows entirely. Only the windowed `lag` approach yields a per-row previous timestamp at the original granularity.

Answer

events.groupBy('user_id').agg(max('ts').alias('prev_ts'))

Answer

events.sortWithinPartitions('user_id', 'ts').withColumn('prev_ts', lead('ts'))

Answer

events.dropDuplicates(['user_id']).withColumnRenamed('ts', 'prev_ts')

Question 9

A data engineer has a batch DataFrame `df` with a `status` string column and wants to keep only rows where `status` equals "active". The engineer wants the filter applied as early as possible in the plan and does not want a shuffle. Which operation best fits this requirement?

Accepted Answer

df.where(col("status") == "active"). Filtering rows by a column predicate is exactly what where() or filter() does, and it is a narrow transformation that runs within each partition without exchanging data. Aggregation, distinct, and sorting all introduce shuffles that the scenario rules out. The where() call also exposes the predicate to Catalyst so it can be pushed into the source when supported, making it both correct and efficient.

Answer

df.select("status").distinct()

Answer

df.groupBy("status").count().filter(col("status") == "active")

Answer

df.orderBy("status").filter(col("status") == "active")

Question 10

A data engineer is writing a PySpark job that must read Parquet data, apply transformations, and then write the result. They want to avoid recomputation if the resulting DataFrame is referenced multiple times in later stages. Which action best accomplishes this?

Accepted Answer

Call df.cache() or df.persist() before the DataFrame is reused. Marking the DataFrame with `cache` or `persist` stores its computed partitions so subsequent actions reuse them rather than replaying the full lineage. Forcing a count, converting to an RDD, or checkpointing without a configured directory either fails to store reusable data or introduces errors and overhead. Caching is the standard mechanism for avoiding recomputation across multiple downstream uses.

Answer

Call df.checkpoint() without setting a checkpoint directory

Answer

Call df.count() to force computation before reuse

Answer

Call df.rdd to convert the DataFrame to an RDD

Question 11

A data engineer has a DataFrame `orders` with columns `order_id`, `customer_id`, and `amount`, and a small lookup DataFrame `tiers` with `customer_id` and `tier`. The engineer wants to attach the tier to every order. Some orders have a customer_id that is not present in tiers, and those orders must still appear with a null tier. Which operation produces this result?

Accepted Answer

orders.join(tiers, on="customer_id", how="left"). The requirement is to keep all orders while enriching them with tier where available, filling null otherwise. A left outer join does precisely this: it preserves the left side unconditionally and matches right-side rows on the key, leaving nulls for misses. Inner join drops unmatched orders, right join preserves the wrong side, and union performs no key-based matching at all.

Answer

orders.join(tiers, on="customer_id", how="inner")

Answer

orders.union(tiers)

Answer

orders.join(tiers, on="customer_id", how="right")

Question 12

A data engineer has two DataFrames: `orders` with columns `order_id` and `customer_id`, and `customers` with columns `customer_id` and `region`. They must produce a result containing only orders whose `customer_id` exists in `customers`, keeping every matching order row exactly once with no customer columns added. Which operation should be used?

Accepted Answer

orders.join(customers, 'customer_id', 'left_semi'). A left semi join keeps only left-side rows that have a matching key on the right, returns just the left columns, and never duplicates rows regardless of duplicate keys on the right. An inner join would add the region column and could duplicate rows, a left anti join inverts the logic, and a union mixes incompatible data. The semi join matches the requirement exactly.

Answer

orders.join(customers, 'customer_id', 'left_anti')

Answer

orders.join(customers, 'customer_id', 'inner')

Answer

orders.union(customers.select('customer_id').withColumnRenamed('customer_id', 'order_id'))

Question 13

A data engineer is building a feature pipeline and needs to add a monotonically increasing integer column `row_num` to a DataFrame `df` that assigns consecutive numbers to rows within each partition of a specified ordering, similar to a window function. Which approach uses the DataFrame API to compute this value?

Accepted Answer

df.withColumn("row_num", row_number().over(Window.orderBy("some_col"))). Consecutive numbering based on a defined ordering is the job of the row_number() window function applied with over(Window.orderBy(...)). It yields one-based consecutive integers according to the window specification, which is what the feature pipeline needs. monotonically_increasing_id() leaves gaps and ignores ordering, a literal constant does not increment, and the RDD zipWithIndex approach collapses to one partition and bypasses DataFrame optimizations.

Answer

df.withColumn("row_num", monotonically_increasing_id())

Answer

df.withColumn("row_num", lit(1).cast("int"))

Answer

df.rdd.zipWithIndex().toDF()

Question 14

A developer has a DataFrame `df` with columns `id` and `score`, and several rows contain null values in `score`. They want a new DataFrame in which rows with a null `score` are removed, keeping only rows where `score` is present. Which single call achieves this?

Accepted Answer

df.na.drop(subset=["score"]). Dropping rows based on nulls in a specific column is done with na.drop and the subset argument, which limits the null check to the named columns. Filling values keeps the rows, filtering for nulls keeps the wrong rows, and dropping a column removes the field entirely, so none of those match the stated goal of discarding incomplete rows.

Answer

df.na.fill(0, subset=["score"])

Answer

df.filter(col("score").isNull())

Answer

df.drop("score")

Question 15

A developer is building a feature that must, for each `customer_id`, concatenate the distinct `product` values from many rows into a single comma-separated string. The result must contain each product only once per customer. Which single approach produces this result?

Accepted Answer

df.groupBy("customer_id").agg(concat_ws(",", collect_set("product")).alias("products")). Producing a distinct comma-separated list per customer requires an aggregation that deduplicates first. collect_set returns a set of unique product values, which concat_ws then joins into one string. collect_list keeps duplicates, and the other approaches either collapse to a single value or reshape the data into columns, none of which yield the required distinct concatenated string.

Answer

df.groupBy("customer_id").agg(concat_ws(",", collect_list("product")).alias("products"))

Answer

df.groupBy("customer_id").pivot("product").agg(count("product"))

Answer

df.select("customer_id", concat_ws(",", "product")).groupBy("customer_id").agg(first("product"))