Databricks-Spark-Assoc Developing DataFrame/DataSet API Applications — 30 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)

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

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'))

Question 11

A data engineer has a PySpark DataFrame `df` with columns `event_id`, `user_id`, and `payload` (a StringType column holding JSON text). The engineer must parse `payload` into a struct column exposing fields `action` and `duration`. Which TWO approaches correctly produce a struct column from the JSON string? (Choose two.)

Accepted Answer

df.selectExpr('payload', "from_json(payload, 'action STRING, duration INT') as parsed"). Both `from_json` used as a column function and the equivalent `from_json` expression inside `selectExpr` parse a JSON string into a struct using a declared schema, keeping the work in the JVM and under Catalyst. The other choices either return only a schema, force slow Python deserialization, or attempt to build a struct from nonexistent columns rather than parsing the JSON text.

Answer

df.withColumn('parsed', struct('action', 'duration'))

Answer

schema_of_json(df.payload)

Answer

df.rdd.map(lambda r: json.loads(r.payload)).toDF()

Question 12

A data engineer has a streaming DataFrame `events` from a Kafka source with columns `user_id` and `event_time`, and must compute a per-user count of events that updates continuously as new micro-batches arrive. Which TWO operations or clauses are required to produce this continuously updated result? (Choose two.)

Accepted Answer

writeStream.format("console").outputMode("update"). A continuously updated per-user count needs an aggregation that groups by user_id and counts events, plus an output mode that re-emits changed rows as new micro-batches arrive. The groupBy().count() provides the aggregate, and writeStream with update output mode delivers the refreshed results. Source offset options, watermarking with deduplication, and caching are supplemental or inapplicable and are not required to achieve the continuously updated behavior.

Answer

df.cache() before calling writeStream

Answer

readStream.format("kafka").option("startingOffsets", "earliest")

Answer

withWatermark("event_time", "10 minutes") followed by dropDuplicates()

Question 13

A developer has a PySpark DataFrame `events` with a `timestamp` column stored as a string in ISO-8601 format (for example, `2024-03-15T08:45:30Z`). They need to extract only the calendar date (year-month-day) into a new column `event_date` of type DateType, without changing the original `timestamp` column. Which single expression accomplishes this?

Accepted Answer

events.withColumn("event_date", to_date(col("timestamp"))). Converting an ISO-8601 string column to a real DateType requires a date-aware parser rather than string slicing or a naive cast. The to_date function interprets the string and returns a DateType, and withColumn preserves the original column while appending the new one. The other approaches either return a string type or fail to parse the ISO-8601 layout, so they do not meet the requirement.

Answer

events.withColumn("event_date", substring(col("timestamp"), 1, 10))

Answer

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

Answer

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

Question 14

A data engineer is working with a DataFrame `df` that has a column `event_time` of type Timestamp and a column `user_id`. The engineer needs to create a new column `session_id` that increments by 1 whenever the gap between consecutive `event_time` values for the same `user_id` exceeds 30 minutes. The data is sorted by `user_id` and `event_time`. Which approach correctly implements this sessionization logic?

Accepted Answer

Use a window partitioned by user_id ordered by event_time, compute lag of event_time, flag rows where the difference exceeds 30 minutes, then use a cumulative sum over the flag to generate session_id.. The correct approach uses a window partitioned by user_id and ordered by event_time, computes the lag of event_time, flags rows where the difference exceeds 30 minutes, and then computes a cumulative sum over that flag to generate session_id. This accurately groups events into sessions based on inactivity gaps.

Answer

Use groupBy('user_id').agg(collect_list('event_time')) and then write a Python UDF to iterate over the list and assign session IDs.

Answer

Use a window partitioned by user_id ordered by event_time, compute the difference between current and previous event_time using lag, and then use a running sum of 1 for each row to generate session_id.

Answer

Use a window partitioned by user_id ordered by event_time, compute the difference between current and previous event_time using lag, and then use a window function to assign a dense_rank based on that difference.

Question 15

A developer must remove duplicate rows from a DataFrame `orders` so that only the most recent order per `customer_id` is kept, based on `order_ts`. Which TWO approaches achieve this correctly? (Choose two.)

Accepted Answer

Sort descending by `order_ts`, then use `dropDuplicates(["customer_id"])`.. A windowed `row_number` filtered to 1 gives a deterministic single latest row per customer with all columns intact, and sorting descending before `dropDuplicates` on the key captures the newest row as the survivor. Both produce one complete order per customer, whereas the tuple-based and aggregate approaches either keep stale rows or discard needed columns.

Answer

Use `dropDuplicates(["customer_id", "order_ts"])` on the DataFrame.

Answer

Use `groupBy("customer_id").agg(max("order_ts").alias("order_ts"))`.

Answer

Use `distinct()` on the DataFrame after selecting `customer_id` and `order_ts`.

Question 16

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.)

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 17

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 18

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 19

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 20

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))