Developing Code for Data Processing using Python and SQL: Databricks Data Engineer Professional Practice Questions

A data engineer is implementing a Pandas UDF that computes a rolling 7-day weighted moving average of revenue for each store_id partition. The function requires access to multiple rows within a group simultaneously. The engineer writes the following code: ``python from pyspark.sql.functions import pandas_udf from pyspark.sql import Window import pandas as pd @pandas_udf('double') def weighted_moving_avg(revenue: pd.Series) -> pd.Series: return revenue.rolling(window=7, min_periods=1).mean() result_df = df.withColumn( 'wma_revenue', weighted_moving_avg(col('revenue')).over( Window.partitionBy('store_id').orderBy('sale_date') ) ) ` When this code runs, it raises: AnalysisException: Pandas UDF does not support 'over' with a Window spec`. What is the correct approach to implement this grouped rolling calculation using Pandas UDFs?

1. AUse a Pandas UDF of type GROUPED_MAP (decorated with @pandas_udf(schema, PandasUDFType.GROUPED_MAP) or the modern applyInPandas pattern). Apply it using df.groupBy('store_id').applyInPandas(weighted_moving_avg_fn, schema=result_schema) where the function receives a full Pandas DataFrame for each store_id group and returns a Pandas DataFrame. This allows applying rolling window logic within each partition using native Pandas operations.
2. BChange the UDF decorator to @pandas_udf('double', PandasUDFType.SCALAR_ITER). Iterator-based Pandas UDFs support window operations because they receive an iterator of batches that correspond to window frames, allowing the rolling computation to be applied across each window's rows.
3. CThe @pandas_udf('double') decorator is correct, but the .over(Window.partitionBy(...)) call must be replaced with .over(Window.partitionBy('store_id').orderBy('sale_date').rowsBetween(-6, 0)) to specify an explicit 7-row frame. Scalar Pandas UDFs support window frames when the frame bounds are explicitly specified using rowsBetween or rangeBetween.
4. DThe solution is to convert the Pandas UDF into a standard Python UDF with @udf('double'). Standard Python UDFs support .over() window expressions because they operate row-by-row and do not require the batching constraints that cause Pandas UDFs to be incompatible with Spark window operations.
5. EPandas UDFs cannot be used for grouped window operations in PySpark. The only correct approach is to collect the DataFrame to the driver with toPandas(), apply the grouped rolling window logic using df.groupby('store_id').apply(lambda g: g.assign(wma=g['revenue'].rolling(7).mean())), and then convert the result back to a Spark DataFrame with spark.createDataFrame().

A data engineer is writing unit tests for a PySpark transformation function apply_revenue_adjustments(df: DataFrame) -> DataFrame that performs type casting, a column computation, and a filter. The engineer wants to test the function locally without connecting to a Databricks cluster, using pytest. The test should verify both the schema of the output DataFrame and the correctness of specific row values. The engineer also wants to use DataFrame.transform() for composing multiple transformation functions in a testable pipeline chain. Which test implementation correctly uses assertSchemaEqual, assertDataFrameEqual, and DataFrame.transform()?

1. A``python def test_revenue_adjustments(spark): input_df = spark.createDataFrame( [(1, '100.5', 'active'), (2, '200.0', 'inactive')], ['id', 'revenue_str', 'status'] ) result = input_df.transform(apply_revenue_adjustments) expected_schema = StructType([ StructField('id', LongType(), True), StructField('revenue', DoubleType(), True), StructField('status', StringType(), True) ]) assertSchemaEqual(result.schema, expected_schema) expected_df = spark.createDataFrame([(1, 100.5, 'active')], ['id', 'revenue', 'status']) assertDataFrameEqual(result, expected_df) ``
2. B``python def test_revenue_adjustments(spark): input_df = spark.createDataFrame( [(1, '100.5', 'active'), (2, '200.0', 'inactive')], ['id', 'revenue_str', 'status'] ) result = apply_revenue_adjustments(input_df) assert result.schema == expected_schema assert result.count() == 1 assert result.first()['revenue'] == 100.5 ``
3. C``python @pytest.mark.parametrize('input_val,expected', [('100.5', 100.5), ('200.0', 200.0)]) def test_revenue_adjustments(spark, input_val, expected): df = spark.createDataFrame([(1, input_val, 'active')], ['id', 'revenue_str', 'status']) result = df.transform(apply_revenue_adjustments) actual = result.select('revenue').first()[0] assert actual == expected ``
4. D``python def test_revenue_adjustments(): input_data = [{'id': 1, 'revenue_str': '100.5', 'status': 'active'}, {'id': 2, 'revenue_str': '200.0', 'status': 'inactive'}] result = apply_revenue_adjustments(input_data) assert result[0]['revenue'] == 100.5 assert len([r for r in result if r['status'] == 'active']) == 1 ``
5. E``python def test_revenue_adjustments(spark): input_df = spark.createDataFrame( [(1, '100.5', 'active'), (2, '200.0', 'inactive')], ['id', 'revenue_str', 'status'] ) result = input_df.transform(apply_revenue_adjustments) expected_df = spark.createDataFrame([(1, 100.5, 'active')], ['id', 'revenue', 'status']) assertDataFrameEqual(result, expected_df, checkRowOrder=False) assertSchemaEqual(result.schema, expected_df.schema, ignoreNullable=True) ``

A data engineering team is configuring a production Databricks Job with the following requirements: (1) the main transformation notebook task requires 64 GB of memory per node due to large in-memory joins; (2) the pipeline must not retry on failure — a failed run should immediately alert the on-call engineer without any automatic re-execution; (3) the job runs in a production environment where the cluster must use Photon-enabled Databricks Runtime and memory-optimized instance types. Which combination of job and cluster configuration settings correctly satisfies all three requirements in a Databricks Asset Bundle YAML?

1. ASet max_retries: -1 in the task configuration to disable all retries. Select a Standard_E64s_v3 (Azure) or r5.2xlarge (AWS) memory-optimized instance type with 64 GB RAM. Set spark_version to a Photon-enabled DBR (e.g., 14.3.x-photon-scala2.12). Use timeout_seconds: 0 at the job level to remove any execution time limit that could interfere with long-running memory-intensive joins.
2. BSet retry_on_timeout: false and max_retries: 1 in the task configuration. The retry_on_timeout: false flag is the correct setting to prevent retries; max_retries: 1 is required as a minimum value because Databricks does not allow max_retries: 0 for production jobs. Select a memory-optimized instance type and set the Spark config spark.executor.memory=64g to explicitly allocate 64 GB per executor.
3. CSet max_retries: 0 in the task configuration to disable automatic retries. Select a memory-optimized instance type for the job cluster worker nodes (e.g., Standard_E64s_v3 on Azure with 432 GB RAM, or r5.16xlarge on AWS with 512 GB RAM). Set spark_version to a Photon-enabled DBR. Configure a job-level email or webhook notification for the on_failure event to alert on-call engineers immediately when the job fails.
4. DSet max_retries: 0 at the job level (not the task level) to apply a zero-retry policy across all tasks. The job-level retry setting overrides any task-level retry configuration. Memory requirements are satisfied by setting spark.driver.memory=64g in the cluster Spark config, which automatically sizes worker memory to match. Photon is enabled by adding photon: true as a top-level cluster property.
5. ESet on_failure: skip_remaining_tasks in the task configuration to prevent downstream tasks from running after a failure, which effectively prevents retries. Set spark.memory.fraction=0.9 to maximize the memory available for joins. Select the largest available instance type and rely on Spark's adaptive memory management to allocate the required 64 GB for joins dynamically.

A data engineering team is implementing a CDC pipeline using APPLY CHANGES INTO in Lakeflow SDP. The source CDC stream sometimes delivers events out of order — a DELETE event for a customer can arrive before the INSERT event for the same customer due to upstream replication lag. The business requirement is that the Silver table always reflects the last known state based on event timestamp, even when events arrive out of order. The team also wants to maintain a full historical audit trail showing all versions of each customer record, not just the current state. Which APPLY CHANGES configuration correctly satisfies both requirements?

1. AAPPLY CHANGES INTO silver_customers FROM STREAM(bronze_cdc) KEYS (customer_id) APPLY AS DELETE WHEN operation = 'DELETE' SEQUENCE BY event_ts; — The SEQUENCE BY event_ts clause automatically handles out-of-order events by buffering all events and sorting by timestamp before applying them, regardless of arrival order.
2. BAPPLY CHANGES INTO silver_customers FROM STREAM(bronze_cdc) KEYS (customer_id) APPLY AS DELETE WHEN operation = 'DELETE' SEQUENCE BY arrival_timestamp COLUMNS * EXCEPT (operation); — Using arrival_timestamp (the Kafka message timestamp or Auto Loader file modification time) instead of the business event timestamp ensures events are always applied in the order they were received, regardless of out-of-order delays in the business event timestamp.
3. CAPPLY CHANGES INTO silver_customers FROM STREAM(bronze_cdc) KEYS (customer_id) APPLY AS DELETE WHEN operation = 'DELETE' SEQUENCE BY event_ts STORED AS SCD TYPE 1 IGNORE NULL UPDATES COLUMNS * EXCEPT (operation); — IGNORE NULL UPDATES prevents out-of-order NULL-valued events from overwriting valid current-state data, and SCD TYPE 1 overwrites older versions while keeping only the current state.
4. DAPPLY CHANGES INTO silver_customers_history FROM STREAM(bronze_cdc) KEYS (customer_id) APPLY AS DELETE WHEN operation = 'DELETE' SEQUENCE BY event_ts STORED AS SCD TYPE 2 COLUMNS * EXCEPT (operation); — SCD Type 2 mode maintains a full historical record of all changes by inserting new rows for each version (with __START_AT and __END_AT columns) rather than overwriting the current row, satisfying the audit trail requirement. SEQUENCE BY event_ts ensures out-of-order events are applied in the correct logical order.
5. EAPPLY CHANGES INTO silver_customers FROM STREAM(bronze_cdc) KEYS (customer_id) APPLY AS DELETE WHEN operation = 'DELETE' SEQUENCE BY event_ts STORED AS SCD TYPE 3 COLUMNS * EXCEPT (operation); — SCD Type 3 stores both the current and previous value of each column in side-by-side column pairs (e.g., current_email, previous_email), providing a full audit trail of all changes in a single row without requiring additional join operations.

A data engineering team is deploying a Databricks Job with a Lakeflow Spark Declarative Pipeline task. The pipeline runs in TRIGGERED mode and is expected to fully process all pending data within 30 minutes under normal conditions. The business SLA requires the pipeline to complete within 45 minutes. The team wants the pipeline to automatically stop and alert if it has not completed within 45 minutes, and they want the pipeline cluster to automatically terminate after completion to avoid idle cluster costs. The team is also aware of a known upstream data quality issue that occasionally causes transient failures, and they want a maximum of 2 automatic retries before the alert is raised. Which DAB job configuration correctly implements all three requirements?

1. ASet timeout_seconds: 2700 (45 minutes) at the pipeline task level to stop the pipeline if it exceeds the SLA. Set max_retries: 2 at the task level to allow up to 2 retries. Pipeline clusters auto-terminate when the job finishes by default — no additional configuration is needed for cluster termination.
2. BSet timeout_seconds: 2700 at the job level (not task level) so the entire job run is terminated if it exceeds 45 minutes total. Set max_retries: 2 at the job level to allow retries of the entire job. Pipeline clusters auto-terminate when the TRIGGERED pipeline completes. Configure a job-level on_failure notification to alert when the job fails after exhausting retries.
3. CSet timeout_seconds: 1800 (30 minutes, matching the expected runtime) to ensure the pipeline is strictly bounded to its normal execution time. Set max_retries: 2 at the task level. Add autotermination_minutes: 10 to the pipeline cluster configuration to terminate the cluster 10 minutes after the pipeline completes, providing a buffer for any post-processing cleanup tasks.
4. DSet timeout_seconds: 2700 at the task level. Set max_retries: 2 with min_retry_interval_millis: 30000 (30 seconds between retries) to avoid immediately hammering the transient failure source. Configure an on_failure job notification. Pipeline clusters (in TRIGGERED mode) auto-terminate when the pipeline run completes, so no explicit auto-termination config is needed.
5. EConfigure the pipeline in CONTINUOUS mode with a timeout_seconds: 2700 at the task level. CONTINUOUS mode automatically retries internally on transient failures without the job needing to retry the task. Set max_retries: 0 at the task level since CONTINUOUS mode handles its own resilience, and configure the on_failure notification at the job level.

A data engineering team is operating a production Lakeflow Spark Declarative Pipeline in TRIGGERED mode that processes financial transaction data from Auto Loader. After a routine deployment, the pipeline fails with the following error in the Event Log: SCHEMA_MISMATCH: Schema of data does not match schema of streaming table: expected 'amount DECIMAL(18,2)' but got 'amount DOUBLE'. The upstream source team has confirmed they changed the amount field type from DECIMAL(18,2) to DOUBLE in a recent data export. The pipeline must resume processing as quickly as possible without data loss. What is the CORRECT remediation sequence?

1. AIn the Lakeflow SDP pipeline settings, change cloudFiles.schemaEvolutionMode from none to addNewColumns. Restart the pipeline. The addNewColumns mode automatically handles type changes (not just new columns) by widening the schema to accommodate both type representations.
2. BUpdate the pipeline's streaming table DDL to change the amount column type from DECIMAL(18,2) to DOUBLE to match the new source schema. Perform a full pipeline reset (clicking 'Full Refresh' in the UI or running databricks pipelines start --full-refresh) to rebuild the streaming table from scratch with the new schema. Resume normal TRIGGERED mode operation.
3. CRun ALTER TABLE silver_transactions ALTER COLUMN amount TYPE DOUBLE on the pipeline's output Delta table directly via a Databricks SQL notebook. Then restart the pipeline without a full reset. The schema change applied directly to the Delta table will resolve the mismatch without requiring a full data reprocessing.
4. DUpdate cloudFiles.schemaEvolutionMode to rescue in the pipeline configuration. In rescue mode, records with the new DOUBLE type for amount will be stored in the _rescued_data column as JSON strings instead of failing. The pipeline resumes immediately and rescued records can be re-processed once the permanent schema fix is deployed.
5. EDelete the pipeline's checkpoint directory in DBFS and restart the pipeline. Deleting the checkpoint forces Auto Loader to re-scan all files from the beginning and re-infer the new schema from the latest data. The pipeline will rebuild the streaming table with the correct DOUBLE type automatically.

A data engineering team needs to apply a complex data masking transformation to a pii_value STRING column in a large PySpark DataFrame. The transformation logic is implemented in a Python function mask_pii(value: str) -> str that calls an external encryption library. The team considers four implementation approaches. The DataFrame has 500 million rows. Which implementation approach offers the BEST performance for this use case, and why?

1. ARegister the function as a standard Python UDF using @udf(returnType=StringType()) and apply it with df.withColumn('masked', mask_pii_udf(col('pii_value'))). Python UDFs are the simplest to implement and Databricks automatically parallelizes them across all executor cores, achieving performance equivalent to native Spark built-in functions for single-column string transformations.
2. BRegister the function as a vectorized UDF using @pandas_udf and apply it with spark.sql() after registering it to the catalog with spark.udf.register('mask_pii', mask_pii_pandas_udf). SQL-invoked UDFs have lower overhead than programmatic DataFrame API UDFs because the SQL query planner can push down the UDF execution to the storage layer, avoiding a full shuffle.
3. CConvert the PySpark DataFrame to a Pandas DataFrame using df.toPandas(), apply the masking function using pandas_df['masked'] = pandas_df['pii_value'].apply(mask_pii), then convert back to a PySpark DataFrame using spark.createDataFrame(pandas_df). This approach leverages Pandas' optimized C-backed string operations, which are faster than Spark's JVM string processing for single-column transformations.
4. DImplement the masking logic as a Spark SQL expression using regexp_replace() and encode()/decode() built-in functions, and apply it via df.withColumn('masked', regexp_replace(col('pii_value'), pattern, replacement)). Built-in Spark SQL functions always outperform any UDF approach because they execute natively in the JVM without any Python process involvement.
5. ERegister the function as a Pandas UDF (vectorized UDF) using @pandas_udf(returnType=StringType()) with the PandasUDFType.SCALAR type (or the modern function annotation approach). Apply it with df.withColumn('masked', mask_pii_pandas_udf(col('pii_value'))). Pandas UDFs use Apache Arrow for serialization, processing data in batches as Pandas Series instead of row-by-row, significantly reducing Python-JVM serialization overhead compared to standard row-at-a-time Python UDFs.

A data engineering team is building a test suite for their production pipeline. A junior engineer proposes using the Databricks notebook's built-in debugger to step through a failing PySpark transformation during interactive development. A senior engineer also wants to ensure the team can run integration tests that validate the full pipeline's end-to-end output against a known-good fixture dataset stored in a Delta table, without requiring a separate test Databricks cluster. Which combination of testing and debugging strategies is MOST appropriate for a production-grade Databricks project?

1. AUse the Databricks notebook debugger for step-through debugging during development, and write integration tests that load the known-good fixture Delta table using spark.read.format('delta').load(fixture_path), apply the full pipeline transformation chain using DataFrame.transform() composition, and then call assertDataFrameEqual(result, expected, checkRowOrder=False) to validate correctness. These integration tests can run in a Databricks job or locally using pyspark with a local SparkSession pointing to DBFS/Volume paths.
2. BUse only the Spark UI's 'SQL / DataFrame' tab to debug failing transformations by examining the physical plan and stage metrics. For integration tests, deploy the pipeline to a dedicated staging Databricks workspace and compare the output Delta table against the production Delta table using EXCEPT SQL queries to identify discrepancies. This environment parity approach is more reliable than local testing for production pipelines.
3. CUse display() calls and notebook cell outputs as the primary debugging mechanism, inserting them at each transformation step to inspect intermediate DataFrames. For integration tests, use spark.sql('SELECT COUNT(*) FROM gold_table') count assertions as the primary correctness check, because full DataFrame comparisons using assertDataFrameEqual are too computationally expensive for production-scale datasets.
4. DUse the Databricks notebook built-in debugger (accessible via the 'Enable Debug' button in the notebook toolbar) to set breakpoints and step through PySpark transformation code line-by-line during development. For integration tests, use assertDataFrameEqual in a pytest test that creates a SparkSession via SparkSession.builder.master('local[*]').getOrCreate() and compares the pipeline output against a small locally-defined fixture DataFrame. Run these tests in CI using pytest without requiring a live Databricks workspace.
5. EUse Python's standard pdb debugger with import pdb; pdb.set_trace() in notebook cells for step-through debugging of PySpark code. For integration tests, mock all Spark DataFrame operations using unittest.mock.patch to avoid the overhead of actually executing Spark plans, and assert that the correct transformation methods were called with the correct arguments.