Data Transformation, Cleansing, and Quality: Databricks Data Engineer Professional Practice Questions

A Delta table sales contains category, product_id, and revenue. A report needs the top 3 products per category by total revenue, but must include ties (for example, if the 3rd and 4th products have the same revenue, both should appear). Which approach best satisfies the requirement?

1. AAggregate revenue per (category, product_id), then compute row_number() over PARTITION BY category ORDER BY total_revenue DESC and filter row_number <= 3.
2. BSkip window functions: GROUP BY category and use ORDER BY total_revenue DESC LIMIT 3 to automatically return the top 3 per category.
3. CAggregate revenue per (category, product_id), then compute dense_rank() over PARTITION BY category ORDER BY total_revenue DESC and filter dense_rank <= 3.
4. DUse rank() and filter rank <= 3; rank() never produces gaps, so it always returns exactly three rows per category.

A PySpark job defines the following window spec for cumulative metrics: w = Window.partitionBy('customer_id').orderBy('event_time') and then computes F.sum('amount').over(w) without calling rowsBetween or rangeBetween. Which window frame is used by default, and what is the correct way to compute a rolling sum over the last 7 rows per customer (row-based, not value/range-based)?

1. ADefault: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING. Rolling 7 rows: keep the same spec; Spark automatically limits to the last 7 rows when an ORDER BY is present.
2. BDefault: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW when an ordering is defined. Rolling 7 rows: w7 = w.rowsBetween(-6, 0) and then F.sum('amount').over(w7).
3. CDefault: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW when an ordering is defined. Rolling 7 rows: w7 = w.rangeBetween(-6, 0) and then F.sum('amount').over(w7).
4. DDefault: no frame is defined; Spark throws an error unless you specify either rowsBetween or rangeBetween. Rolling 7 rows: w7 = w.rowsBetween(-7, -1).

A data engineer runs a nightly MERGE INTO target USING updates ON target.id = updates.id that performs WHEN MATCHED THEN UPDATE and WHEN NOT MATCHED THEN INSERT. The job intermittently fails with an error stating that the MERGE matched multiple source rows for a single target row and cannot deterministically update it. The updates source is a raw CDC feed that can contain several changes for the same id within one batch. Which fix correctly resolves the non-deterministic MERGE while applying only the most recent change per id?

1. AChange the join condition to ON target.id = updates.id AND updates.is_latest = true and rely on Delta to ignore non-latest duplicates; the extra predicate guarantees only one source row per target row participates in the match without pre-aggregating the source.
2. BAdd WHEN MATCHED AND updates.op = 'U' THEN UPDATE plus a second WHEN MATCHED AND updates.op = 'D' THEN DELETE; splitting the matched clause by operation type makes the multi-match deterministic because each matched row now falls into exactly one clause.
3. CSet spark.databricks.delta.merge.repartitionBeforeWrite.enabled to true, which repartitions the source by id so duplicate source rows for the same id are processed by a single task and automatically collapsed into the last write.
4. DBefore the MERGE, collapse the source to one row per id by keeping the latest change (e.g., a row_number() window on id ordered by the change timestamp desc, filtered to 1), then MERGE the deduplicated source so each target row matches at most one source row.
5. EWrap the MERGE in SET spark.databricks.delta.merge.enableLowShuffle = true; low-shuffle merge tolerates multiple matching source rows by applying them in ingestion order, so the multi-match error no longer occurs.

A data engineer must transform a long-format sales(region, product_category, amount) table into a wide report with one row per region and one summed-amount column per product_category. New product categories are added frequently, and the engineer wants the transformation to include every category present in the data at run time without editing the query each time a category is added. Which Spark approach produces the correct dynamic wide output?

1. AUse df.groupBy('region').rollup('product_category').sum('amount'); rollup produces one subtotal column per distinct category value, giving the same wide layout as a pivot while also computing grand totals.
2. BUse df.groupBy('region').pivot('product_category', ['electronics','apparel','home']).sum('amount') with the full category list hardcoded; an explicit list is required because pivot cannot determine the columns otherwise, and unlisted categories are summed into an other column.
3. CUse df.groupBy('region', 'product_category').agg(sum('amount')) and then collect() the distinct categories on the driver to build the wide schema with a Python loop of withColumn calls, because pivot cannot create columns dynamically.
4. DUse a SQL GROUP BY region with a CASE WHEN product_category = ... sum for each category; this is the only way to guarantee all categories appear because CASE expressions are evaluated per row regardless of new categories.
5. EUse df.groupBy('region').pivot('product_category').sum('amount'); calling pivot without an explicit value list triggers a distinct scan of product_category, so all categories present at run time become columns automatically.

A data engineer must deduplicate a Delta table of change events so that, for each account_id, only the record with the latest event_ts is kept; if two records share the same account_id and event_ts, the one with the higher ingest_seq wins, and exactly one row per account_id must remain. The engineer wants a single-pass, deterministic transformation. Which approach is correct and deterministic?

1. AGroup by account_id and select max_by(struct(event_ts, ingest_seq), event_ts); max_by on event_ts alone returns the latest event, and ties are broken arbitrarily by Spark, which is acceptable because tied timestamps represent the same logical record.
2. BDefine w = Window.partitionBy('account_id').orderBy(col('event_ts').desc(), col('ingest_seq').desc()), add row_number().over(w) as rn, and filter rn == 1; the compound ordering makes the winner deterministic and guarantees exactly one row per account_id.
3. CDefine w = Window.partitionBy('account_id').orderBy(col('event_ts').desc()), add rank().over(w) as rk, and filter rk == 1; rank returns the top record per account and resolves ties by keeping all tied rows so no data is lost.
4. DUse dropDuplicates(['account_id']) after sorting the DataFrame globally by event_ts descending and ingest_seq descending; the prior global sort guarantees dropDuplicates keeps the first (latest) row it encounters per account_id.
5. EUse df.groupBy('account_id').agg(first('event_ts', ignorenulls=True)) with the DataFrame pre-sorted by ingest_seq; first after sorting returns the latest event deterministically because Spark preserves sort order through the aggregation.

A data engineer builds a Lakeflow Spark Declarative Pipeline (Delta Live Tables) that loads a Silver transactions table. The requirement is: rows that fail a critical data-quality rule (amount > 0) must NOT enter the Silver table, but they must also NOT be lost — they must be routed to a separate quarantine table for investigation, and the pipeline must keep running rather than aborting the update. Which implementation meets all of these requirements?

1. AApply @dlt.expect_or_drop('valid_amount', 'amount > 0') on the Silver table to remove failing rows, and define a second table that reads the same source with the inverted predicate (amount <= 0 OR amount IS NULL) as its own expectation-free quarantine table, so valid rows flow to Silver and invalid rows are retained separately while the pipeline continues.
2. BApply @dlt.expect_or_fail('valid_amount', 'amount > 0') to the Silver table so that failing rows halt the update immediately; then re-run the pipeline pointing at only the failed rows to populate a quarantine table from the pipeline event log.
3. CApply @dlt.expect('valid_amount', 'amount > 0') to the Silver table; the warn-level expectation keeps the row out of Silver and Databricks automatically writes every warned row to a system-managed _quarantine table with no additional code.
4. DAdd a CONSTRAINT valid_amount CHECK (amount > 0) ON VIOLATION DROP ROW to the Silver table definition and enable pipelines.quarantine.autoCapture, a pipeline configuration flag that mirrors every dropped row into a quarantine Delta table automatically.
5. EUse @dlt.expect_all_or_drop with the single rule amount > 0; because expect_all_or_drop both drops and captures, the dropped rows are emitted as a change-data-feed stream that a downstream table reads to materialize the quarantine table without a second source scan.

A data engineer writes unit tests for a transformation function enrich(df) that adds derived columns. The test must assert that the transformed DataFrame equals an expected DataFrame regardless of row order, that the two schemas match exactly (names, types, nullability), and it should run on a local test session without a full production cluster. Which testing approach uses the built-in PySpark utilities correctly?

1. ACall assertDataFrameEqual(actual, expected) from pyspark.testing, which compares both data and schema and, by default, treats the DataFrames as equal irrespective of row ordering; optionally pair it with assertSchemaEqual(actual.schema, expected.schema) for an explicit schema-only check.
2. BCollect both DataFrames to Python lists and assert sorted(actual.collect()) == sorted(expected.collect()); this is required because assertDataFrameEqual compares only schemas, not row data, so data equality must be checked manually.
3. CUse actual.exceptAll(expected).count() == 0 as the sole assertion; exceptAll handles both schema and data comparison including nullability, so no separate schema assertion is needed and row order is inherently ignored.
4. DUse assertDataFrameEqual(actual, expected, checkRowOrder=True) as the default call, because the utility fails unless rows are in identical order; to ignore order you must first .orderBy() both DataFrames identically since the function has no order-insensitive mode.
5. ERegister both DataFrames as temp views and run SELECT * FROM actual MINUS SELECT * FROM expected; an empty result proves equality of data, schema, and nullability, which is the officially recommended PySpark unit-test pattern.

A classic job uses Auto Loader to ingest JSON files where new fields can appear without notice and some fields occasionally arrive with type mismatches. The pipeline must keep running (no manual restarts), and unexpected fields must not be silently dropped. Which configuration best satisfies these requirements while minimizing custom state handling?

1. AUse Auto Loader with cloudFiles.schemaEvolutionMode='rescue' (and a persistent cloudFiles.schemaLocation + checkpointLocation). Unexpected columns and type/case mismatches are captured in the rescued data column (for example _rescued_data) so ingestion continues without failing on new columns.
2. BUse cloudFiles.schemaEvolutionMode='addNewColumns' and disable checkpoints. This prevents schema-related failures and guarantees that new columns are written immediately to the target table.
3. CProvide an explicit schema and set cloudFiles.schemaEvolutionMode='addNewColumns'. Auto Loader will automatically evolve the schema and never fail when new columns arrive.
4. DRely on badRecordsPath to capture all unexpected columns and type mismatches; the rescued data column is only for malformed JSON records.