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

Question 1

A developer must persist a DataFrame `clean_df` to Parquet while partitioning the output on disk by the `region` column and controlling the number of output files per partition directory. Which TWO statements about using `clean_df.write` are correct? (Choose two.)

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

Accepted Answer

Calling .partitionBy("region") on the DataFrameWriter creates one directory per distinct region value in the output path.. The writer's partitionBy method only shapes the on-disk directory hierarchy, creating one folder per distinct value of the chosen column, while the actual file count inside each folder is determined by the DataFrame's partition count at write time. The remaining statements misattribute sorting, stale-directory cleanup, or automatic coalescing to partitionBy, none of which it provides.

Answer

Setting the write mode to overwrite with partitionBy guarantees that stale region directories from previous writes are removed.

Answer

partitionBy("region") will automatically coalesce the DataFrame so that exactly one file is produced per region.

Answer

Adding .bucketBy(4, "region") before the write will sort records within each region directory for faster range scans.

Question 2

A data engineer is working with a PySpark DataFrame `df` that contains a column `tags` of type ArrayType(StringType). The engineer needs to create a new column `tag_count` that contains the number of elements in each array, and also filter out rows where the array is empty or null. Which TWO of the following approaches correctly achieve both the transformation and filtering? (Choose two.)

Accepted Answer

df = df.withColumn('tag_count', size(col('tags'))).filter(size(col('tags')) > 0). The correct approaches use the size function to compute the array length and then filter out null and empty arrays. One option relies on size(null) returning -1 to filter nulls implicitly, while the other explicitly checks for non-null. Both are valid. The other options misuse aggregate functions or string functions on arrays.

Answer

df = df.withColumn('tag_count', size(col('tags'))).filter(length(col('tags')) > 0)

Answer

df = df.withColumn('tag_count', size(col('tags'))).filter(array_contains(col('tags'), null))

Answer

df = df.withColumn('tag_count', count(col('tags'))).filter(col('tags').isNotNull() & (size(col('tags')) > 0))

Question 3

A data engineer has a DataFrame `df` with columns `product_id` and `price`. They need to create a new DataFrame that contains only the `product_id` and a new column `price_with_tax` calculated as `price * 1.1`. Which code snippet correctly produces this result?

Accepted Answer

df.select('product_id', (col('price') * 1.1).alias('price_with_tax')). Using select with an aliased expression directly produces a DataFrame with only the desired columns and the computed column. The other options either are less efficient, lack proper aliasing, or retain extra columns. The select method is the most straightforward and correct for this transformation.

Answer

df.withColumn('price_with_tax', col('price') * 1.1).select('product_id', 'price_with_tax')

Answer

df.select('product_id', col('price') * 1.1)

Answer

df.withColumn('price_with_tax', col('price') * 1.1).drop('price')

Question 4

A developer has a DataFrame `df` with a nested struct column `address` containing fields `city`, `state`, and `zip`. They need to expose `city` and `zip` as top-level columns. Which TWO approaches correctly achieve this using the DataFrame API? (Choose two.)

Accepted Answer

df.select("address.city", "address.zip"). Both the string-based dot notation and the col-based dot notation with alias correctly project nested struct fields to top-level columns without changing row count or dropping unrelated fields. The remaining approaches either use explode or getItem on a struct, which is invalid, or drop the entire struct and lose the zip field.

Answer

df.withColumn("city", df.address.city).drop("address")

Answer

df.select(explode("address").alias("city", "state", "zip"))

Answer

df.select(col("address").getItem("city"))

Question 5

A data engineer is using a PySpark DataFrame `df` with a column `category` and a column `value`. They need to compute the percentage of total `value` for each `category`, rounded to two decimal places. The total sum of `value` across all rows must be computed over the entire DataFrame, not per category. Which code snippet correctly achieves this?

Accepted Answer

total = df.agg(sum('value')).collect()[0][0]; df = df.withColumn('pct', round(col('value') / total * 100, 2)). The correct approach computes the global sum using an aggregation and collect, then uses withColumn to calculate the percentage for each row. This row-level percentage can then be summed or averaged by category as needed. The other options incorrectly use window functions partitioned by category or attempt invalid nested aggregations.

Answer

df = df.groupBy('category').agg((sum('value') / df.agg(sum('value')).collect()[0][0] * 100).alias('pct'))

Answer

df = df.withColumn('pct', round(col('value') / sum('value').over(Window.partitionBy('category')) * 100, 2))

Answer

df = df.withColumn('pct', round(col('value') / df.select(sum('value')).first()[0] * 100, 2))

Question 6

A data engineer has a PySpark DataFrame `events` with a `timestamp` column stored as a string in the format `yyyy-MM-dd HH:mm:ss`. They need to add a new column `event_date` that contains only the date part as a `date` type, and they want to avoid any implicit string-to-date casting that could produce nulls on malformed rows. Which approach correctly creates `event_date` while preserving a strict parse behavior?

Accepted Answer

events.withColumn('event_date', to_date(col('timestamp'), 'yyyy-MM-dd HH:mm:ss')). Using `to_date` with an explicit format string is the correct way to convert a known string layout into a `date` column while keeping parse behavior predictable. The explicit pattern removes dependence on session defaults, and non-matching rows become null instead of raising errors. Other approaches either return a string, rely on implicit casting rules, or assume a fixed character offset that may not hold.

Answer

events.withColumn('event_date', substring(col('timestamp'), 1, 10).cast('date'))

Answer

events.withColumn('event_date', col('timestamp').cast('date'))

Answer

events.withColumn('event_date', date_format(col('timestamp'), 'yyyy-MM-dd'))

Question 7

A developer is working with a PySpark DataFrame `transactions` that has columns `account_id`, `txn_date` (DateType), and `amount` (DoubleType). They must produce a new DataFrame containing only the rows where `amount` is strictly greater than 500.00 and `txn_date` is on or after 2024-01-01. The predicate must be pushed down to the underlying Parquet scan to minimize I/O. Which code snippet accomplishes this?

Accepted Answer

transactions.filter((col('amount') > 500.0) & (col('txn_date') >= lit('2024-01-01'))). The requirement is a native Column predicate over base-table columns, which Catalyst can push down to the Parquet reader so that only matching row groups are scanned. The snippet using `filter` with `col` expressions and `lit` for the date literal satisfies this: it stays entirely inside Catalyst, keeps the scan pruned, and avoids Python serialization. Lambda-based filters and RDD conversions break pushdown, and unquoted date strings in SQL predicates change semantics.

Answer

transactions.where(lambda r: r.amount > 500.0 and r.txn_date >= '2024-01-01')

Answer

transactions.select('account_id', 'txn_date', 'amount').filter('amount > 500.0 and txn_date >= 2024-01-01')

Answer

transactions.rdd.filter(lambda r: r['amount'] > 500.0 and r['txn_date'] >= datetime.date(2024, 1, 1)).toDF()

Question 8

You have a large DataFrame containing user transaction logs. You need to read this data and immediately repartition it by user_id to optimize downstream filtering operations. Which DataFrame API method should you use?

Accepted Answer

df.repartition('user_id'). The repartition method creates a new set of partitions across the cluster network, which helps distribute data evenly to prevent skew. This is a critical transformation in Spark to optimize shuffle performance for downstream queries and aggregations by ensuring balanced workloads across executors.

Answer

df.coalesce('user_id')

Answer

df.partitionBy('user_id')

Answer

df.shuffle('user_id')

Question 9

You need to add a new calculated column 'discounted_price' to an existing DataFrame 'df' by multiplying 'price' by 0.9. Which DataFrame transformation accomplishes this correctly?

Accepted Answer

df.withColumn('discounted_price', col('price') * 0.9). Adding columns immutably is a core pattern in Spark DataFrame operations. Using the withColumn method creates a new DataFrame reference with the added transformation while leaving the original dataset unchanged, adhering to functional programming principles required for distributed data processing.

Answer

df.addColumn('discounted_price', col('price') * 0.9)

Answer

df.update('discounted_price', df.price * 0.9)

Answer

df.select(col('*'), col('price') * 0.9 as 'discounted_price')

Question 10

A data engineer needs to join two large DataFrames, `sales` and `products`, on the `product_id` column. The `products` DataFrame is extremely small and fits entirely in a single executor's memory. To optimize performance and avoid a costly shuffle join across the network, which strategy should be applied using the Spark DataFrame API?

Accepted Answer

Call `sales.join(broadcast(products), "product_id")` to explicitly push the small DataFrame to all worker nodes.. Broadcasting small datasets avoids expensive network shuffles by copying the entire small DataFrame to all worker nodes. This optimization drastically improves query performance for star schemas and dimension table lookups. Using the broadcast function explicitly guarantees that the Catalyst optimizer chooses a broadcast hash join instead of a sort-merge join.

Answer

Partition both DataFrames explicitly by `product_id` using `repartition(col("product_id"))` prior to joining.

Answer

Increase the shuffle partition count configuration via `spark.sql.shuffle.partitions` to a higher value.

Answer

Cache the `sales` DataFrame in memory before executing the standard inner join operation.

Question 11

You are processing a large, highly skewed PySpark DataFrame in Databricks and want to optimize a forthcoming join operation against a small lookup dimension table. Which TWO strategies are valid and effective DataFrame API techniques to optimize this join performance? (Choose TWO)

Accepted Answer

Use the broadcast() function on the small dimension table to distribute it to all worker nodes and avoid a shuffled hash join.. Optimizing joins involving skewed datasets requires avoiding shuffles for small tables or breaking up skewed keys. Broadcasting the small table eliminates the shuffle phase entirely, while salting keys distributes skewed join keys across multiple tasks, preventing memory bottlenecks on single executors during heavy shuffle exchanges.

Answer

Apply a broadcast join hint directly to the large skewed fact table to force executor nodes to cache its partitions in memory.

Answer

Increase the spark.sql.shuffle.partitions configuration to an extremely high number like 10000 to eliminate data skew entirely.

Answer

Convert the large DataFrame into a local Pandas DataFrame using toPandas() to perform the join operations locally on the driver node.

Question 12

A data engineer has a PySpark DataFrame `events` with columns `user_id`, `event_time` (timestamp), and `payload` (string). They must produce a new DataFrame where each row is enriched with the `payload` value from the user's immediately preceding event, ordered by `event_time`, without collapsing any rows. Which TWO approaches accomplish this? (Choose two.)

Accepted Answer

Create a Window partitioned by `user_id` and ordered by `event_time`, then use `lag('payload', 1).over(windowSpec)` inside `withColumn`.. Enriching rows with a prior value without collapsing requires a Window function. Partitioning by `user_id` and ordering by `event_time` ensures the 'previous' row is the same user's immediately earlier event, while `lag` returns the payload from that row and preserves row count. Both `lag(...).over(windowSpec)` and the inline `Window.partitionBy(...).orderBy(...)` expression produce identical semantics; the alternatives either collapse rows, use invalid syntax, or lose per-user ordering.

Answer

Use `df.orderBy('event_time').rdd.zipWithIndex()` and manually look up the previous index per user in a driver-side dictionary.

Answer

Apply `window('user_id', 'event_time')` with `lag('payload')` and rely on Spark to fill nulls for the first event.

Answer

Call `groupBy('user_id').agg(collect_list('payload'))` and then explode the resulting list back into rows.

Question 13

A developer has a PySpark DataFrame `df` with columns `order_id`, `customer_id`, and `order_ts` (timestamp). They need to return only the most recent order per customer, keeping all original columns, and they want to avoid a self-join or a manual sort-then-dropDuplicates approach. Which DataFrame operation should they use?

Accepted Answer

Use a Window partitioned by `customer_id` ordered by `order_ts` descending, add `row_number()`, then filter for row number equal to 1. Selecting the latest row per group while preserving all columns is exactly what a Window with `row_number()` solves. Partitioning by the grouping key and ordering by the timestamp descending makes the most recent record rank 1, and filtering on that rank returns complete rows. Aggregations drop columns, global limits collapse groups, and dropDuplicates cannot express ordering.

Answer

df.dropDuplicates(["customer_id"])

Answer

df.groupBy("customer_id").max("order_ts")

Answer

df.orderBy("order_ts", ascending=False).limit(1)

Question 14

A developer is building a PySpark job that reads a Parquet dataset with 2,000 files into `df`. They call `df.cache()` and then execute three separate actions in the same session. The Spark UI shows the Parquet files are read from storage three times, and the cache never appears in the Storage tab. What is the most likely cause?

Accepted Answer

`cache()` was applied to `df` but a variable reassignment such as `df = df.withColumn(...)` created a new logical plan that no longer references the cached node.. Caching in Spark marks a specific logical plan node as persistent. Any subsequent transformation that derives a new DataFrame from it produces a different plan that does not include the cached node, so actions against the new DataFrame bypass the cache entirely. The tell-tale signs here are repeated Parquet reads and an empty Storage tab. To fix this, either cache after the final transformation or reuse the exact same cached DataFrame reference across actions.

Answer

`cache()` only takes effect after `persist(StorageLevel.MEMORY_ONLY)` is called, so the default call does nothing.

Answer

`cache()` stores data in memory only, and the dataset is too large, so Spark silently drops the cached blocks.

Answer

The Parquet files are stored in a format that Spark cannot cache, so `cache()` is a no-op for Parquet sources.

Question 15

A developer has a DataFrame `events` with columns `user_id`, `event_type`, and `payload` (a JSON string). They need to extract the `device` field from `payload` into a new column without changing the other columns or the row count. Which approach is correct?

Accepted Answer

events.withColumn("device", get_json_object("payload", "$.device")). Extracting a single scalar from a JSON string column is exactly what `get_json_object` does, and `withColumn` adds the result without altering other columns or row counts. Parsing the whole payload with `from_json` is unnecessary when only one field is needed, and any use of `explode` or RDD conversion changes the shape of the DataFrame.

Answer

events.withColumn("device", explode(from_json("payload", schema)))

Answer

events.select("user_id", "event_type", from_json("payload", schema).getField("device"))

Answer

events.rdd.map(lambda r: r.payload["device"]).toDF()