16 Sep, 2026
0 Comments
85 Mins Read

PySpark & Databricks Interview Questions and Answers

You have a job where 199 tasks finish in 2 minutes but 1 task takes 40 minutes. How would you diagnose and fix it?

This is a classic data skew / straggler task problem.

If 199 tasks complete in 2 minutes and one task takes 40 minutes, I would first determine whether the slow task is processing disproportionately more data or whether it has a different execution characteristic.

First, inspect the Spark UI

I would check:
- Stage details
- Task duration
- Input size per task
- Output size per task
- Records processed per task
- Shuffle read/write
- Spill to memory/disk
- GC time
- Executor running the slow task
For example:
text
```
Task 0    500 MB    2 min
Task 1    480 MB    2 min
...
Task 199  490 MB    2 min
Task 200  80 GB     40 min
```
That strongly indicates partition/data skew.

Identify the skewed key

Suppose I’m joining:
text
```
employee_id
```
I would check the distribution:
sql
```
SELECT employee_id, COUNT(*) AS cnt
FROM source
GROUP BY employee_id
ORDER BY cnt DESC
LIMIT 20;
```
If one employee/customer/account has millions or billions of records while others have very few, that key is causing skew.

Check whether the problem is actually skew

Not every slow task means skew.

I would also investigate:
- Uneven input file sizes
- Too few partitions
- Small-file problems
- Expensive UDF
- Data type conversion
- Excessive shuffle
- Executor GC
- Disk spilling
- Network bottleneck
Fix the root cause

Depending on the cause:

Solution A — Repartition

If partitioning is poor:
python
```
df = df.repartition(500, "employee_id")
```
But this alone may not solve severe key skew because all records belonging to one key can still end up together.

Solution B — Salting

For a heavily skewed key:
python
```
from pyspark.sql.functions import rand, floor

df = df.withColumn(
    "salt",
    floor(rand() * 10)
)
```
Then distribute the large key across multiple partitions.

For a join, the other side may also need corresponding salted rows.

Solution C — Broadcast the small table

If one side is small enough:
python
```
from pyspark.sql.functions import broadcast

result = large_df.join(
    broadcast(small_df),
    "employee_id"
)
```
This avoids a large shuffle.

Solution D — Enable Adaptive Query Execution

Spark AQE can dynamically handle certain skewed joins:
text
```
spark.sql.adaptive.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
```
Interview conclusion

“I would not immediately increase cluster size. One task running 40 minutes while 199 finish in 2 minutes strongly suggests a straggler, often caused by data skew. I would verify it through Spark UI by comparing partition sizes, shuffle read, spill and records processed. Then I would address the root cause using AQE/skew join handling, repartitioning, salting, or broadcasting where appropriate.”
Design a schema drift detection framework. What drift types does it catch, and what’s the remediation path?

This question has two parts:

Design the framework.
Apply it to the renamed-column scenario.

Part A — Requirement gathering

Before designing the solution, I would clarify:

Source questions
- What type of source is it?
- Batch or streaming?
- Is schema supplied by the source or inferred?
- How frequently can the schema change?
- Are changes backward compatible?
- Who owns the source schema?
Business questions
- Which changes are acceptable?
- Can columns be added?
- Can columns be removed?
- Can data types change?
- Are renames allowed?
- Is breaking the pipeline acceptable?
Operational questions
- How quickly must drift be detected?
- Should the pipeline stop or continue?
- Who receives alerts?
- Do we need automatic remediation?
- Do we need historical backfill?
Part B — Drift types

I would detect at least:

Column addition

Example:
text
```
employee_id
name
salary
department
```
becomes:
text
```
employee_id
name
salary
department
email
```
Usually backward compatible.

Action:
- Register the new column.
- Update downstream schema if allowed.
- Alert data owners.
Column deletion

Example:
text
```
salary
```
disappears.

Potentially breaking.

Action:
- Alert.
- Quarantine/stop downstream processing depending on criticality.
- Confirm with source owner.
Column rename

Example:
text
```
employee_name
```
becomes:
text
```
name
```
This is dangerous because many systems interpret it as:

old column removed
+
new column added

rather than a rename.

Action:
- Detect based on metadata plus similarity.
- Require explicit mapping.
- Backfill historical data.
Data type change

Example:
text
```
salary: INT
```
becomes:
text
```
salary: STRING
```
Action:
- Validate compatibility.
- Cast if safe.
- Otherwise quarantine.
Nullability change

Example:
text
```
employee_id NOT NULL
```
becomes nullable.

Action:
- Validate downstream expectations.
- Alert if business-critical.
Constraint/key changes

Examples:
- Primary key removed
- Unique key changed
- Partition column changed
These can have significant downstream impact.

Framework architecture

I would maintain a schema registry / schema history table.

Example:
text
```
schema_registry
----------------------------
source_name
table_name
version
column_name
data_type
nullable
ordinal_position
first_seen
last_seen
```
Every ingestion run would:
text
```
Source
   ↓
Extract schema
   ↓
Normalize schema
   ↓
Compare with registered schema
   ↓
Classify drift
   ↓
Policy engine
   ↓
Alert / Accept / Quarantine / Transform
   ↓
Bronze
   ↓
Silver
```
Apply the scenario

Source silently renamed employee_name to name, and Bronze contains both names across partitions.

This is dangerous because Bronze now contains:
text
```
Partition 1:
employee_id
employee_name

Partition 2:
employee_id
name
```
If we simply merge them, Silver can get:
text
```
employee_name = NULL
```
for newer records or:
text
```
name = NULL
```
for older records.

Detect it before Silver

At ingestion time, compare the incoming schema against the schema registry.

If previous schema was:
text
```
employee_id
employee_name
salary
```
and new schema is:
text
```
employee_id
name
salary
```
the framework identifies:
text
```
employee_name → removed
name → added
```
Then use additional signals to classify it as a likely rename:
- Similar data type
- Similar nullability
- Similar position
- Similar values
- Column-name similarity
- Same source field metadata
- Same semantic/business meaning if available
But I would not automatically assume every add+drop is a rename.

Fix historical Bronze data

Create a canonical Silver column:
text
```
employee_name
```
Then normalize both forms:
sql
```
SELECT
    employee_id,
    COALESCE(employee_name, name) AS employee_name,
    salary
FROM bronze
```
For historical Bronze data, I would preferably not destroy the raw Bronze data.

Instead:
text
```
Raw Bronze
     ↓
Normalization
     ↓
Canonical Bronze/Silver
```
If governance permits rewriting the curated layer, backfill historical partitions:
text
```
old employee_name → employee_name
new name           → employee_name
```
Then validate:
sql
```
SELECT COUNT(*)
FROM silver
WHERE employee_name IS NULL;
```
and compare record counts before and after.

Interview conclusion

“I would preserve raw Bronze, introduce a canonical schema in the curated layer, create an explicit rename mapping, backfill historical partitions using COALESCE or a controlled transformation, and add schema-contract validation so the same drift cannot silently reach Silver again.”
Streaming job crashes mid-micro-batch after writing output but before checkpoint commits. What happens on restart?

This is an important Spark Structured Streaming + Delta Lake question.

Suppose a micro-batch is:
text
```
Batch 100
```
The job:
- Reads input.
- Processes records.
- Writes Delta output.
- Crashes.
- Checkpoint for batch 100 hasn’t committed.
On restart, Spark sees that batch 100 was not successfully committed in the checkpoint.

Therefore it can attempt to process batch 100 again.

Why doesn’t this necessarily create duplicates?

Delta Lake provides transactional writes.

The sink operation is committed atomically through the Delta transaction log.

Structured Streaming tracks progress through the checkpoint.

The combination gives the intended exactly-once processing semantics for supported source/sink configurations.

The important distinction is:

Spark may retry the batch, but the Delta transaction mechanism prevents an already-committed batch from being blindly duplicated when the streaming sink uses the appropriate transaction semantics.

Conceptually:
text
```
Input
  ↓
Batch 100
  ↓
Transform
  ↓
Delta transaction
  ↓
_commit_100
  ↓
Checkpoint
```
If failure occurs around the commit boundary, restart uses the checkpoint and Delta transaction information to achieve idempotent/transactional behavior.

Important interview nuance

I would not say:

“Exactly once means every external side effect is exactly once.”

That’s incorrect.

For example:
text
```
write_to_database()
send_email()
call_rest_api()
```
Delta cannot make arbitrary external side effects exactly once.

Exactly-once guarantees depend on the source, checkpointing, sink, and transaction semantics.

Practical prevention

I would ensure:
- Checkpoint is stored on durable storage.
- Checkpoint path is unique per streaming query.
- Don’t manually delete checkpoints.
- Don’t reuse one checkpoint for different query logic without understanding state compatibility.
- Delta is used as the transactional sink.
- External sinks are designed to be idempotent.
You ran VACUUM with 0-hour retention and a long-running streaming read fails with “file not found.” Why?

This is a classic Delta Lake retention problem.

Delta tables have:

text

Data files + Delta transaction log

Older versions may still be referenced by a long-running reader.

Suppose:

text

Streaming reader ↓ Version 100

Meanwhile the table is updated:

text

100 → 101 → 102 → 103

Then:

text

VACUUM 0 HOURS

removes files that are no longer considered current and are outside the retention window.

If the streaming reader still needs one of those physical files, it can receive:

text

FileNotFoundException

Why 0 hours is dangerous

The retention period protects readers and other processes that may still require older files.

A long-running streaming job can lag behind table maintenance.

So:

text

VACUUM retention

must account for:

Long-running queries

Streaming jobs

Time-travel requirements

Delayed jobs

Recovery processes

Prevention

Don’t use zero-hour retention casually

Use an appropriate retention period.

Align retention with maximum reader lag

For example, if a streaming workload could be unavailable for 24 hours, retention needs to accommodate that operational window plus safety margin.

Monitor streaming lag

Track:

text

latest source version vs streaming processed version

Protect against unsafe VACUUM

In Databricks/Delta environments, safety checks exist specifically to prevent dangerous retention settings.

Don’t disable them simply to save storage without understanding the consequences.

Interview answer

“VACUUM physically deletes old files; it doesn’t just clean metadata. A long-running streaming reader may still need those files. If I remove them with zero-hour retention, the reader can fail with FileNotFoundException. I would use an appropriate retention policy based on the maximum reader lag, monitor streaming lag, and avoid unsafe VACUUM overrides.”
Two source records for the same employee arrive in the same batch with different attributes. How does SCD Type 2 MERGE handle this?
This is a very important point:

Don’t blindly MERGE the raw source batch.

Suppose the batch contains:

text

employee_id | department ------------|----------- 101 | IT 101 | Finance

A standard SCD2 MERGE expects a deterministic source.

If both records match the same target row, the merge can become ambiguous and may fail or produce an incorrect result depending on implementation/version.

Step 1 — Deduplicate the source

Define the business rule.

For example:

text

employee_id + effective_timestamp

determines which event is latest.

Then:

python

from pyspark.sql.window import Window from pyspark.sql.functions import row_number, col window = Window.partitionBy("employee_id") \ .orderBy(col("effective_ts").desc()) deduped = ( source .withColumn("rn", row_number().over(window)) .filter("rn = 1") )

Now only one current record remains per employee.

But there is another issue

Suppose these are genuinely two historical changes:

text

10:00 → IT 10:05 → Finance

If I simply retain the latest row, I lose the IT change.

Therefore, for a true CDC/event stream, I should process the changes in event order, not simply deduplicate them.

SCD Type 2 structure

Typical target:

text

employee_id department effective_from effective_to is_current

Existing:

text

101 | HR | Jan 1 | NULL | true

New:

text

101 | IT | Jan 10 101 | Finance | Jan 15

Expected history:

text

101 | HR | Jan 1 | Jan 10 | false 101 | IT | Jan 10 | Jan 15 | false 101 | Finance | Jan 15 | NULL | true

Processing strategy

I would:

Validate source uniqueness.

Order multiple changes by reliable event timestamp/sequence.

Detect actual attribute changes.

Close the existing current record.

Insert new versions.

Ensure exactly one current record exists.

Validation:

sql

SELECT employee_id FROM employee_scd WHERE is_current = true GROUP BY employee_id HAVING COUNT(*) > 1;

Expected result:

text

0 rows

Interview conclusion

“An SCD2 MERGE should receive a deterministic source. If two records for the same business key arrive in the same batch, I first determine whether they’re duplicates or legitimate sequential changes. For duplicates, I deduplicate using a sequence/timestamp. For legitimate changes, I process them in event order so every valid historical version is preserved.”
A production job ran fine for months and suddenly starts spilling to disk. What changed?
I would not immediately increase executor memory.

Spilling means Spark couldn’t keep the working data in memory during operations such as:

Sort

Shuffle

Aggregation

Join

The key question is:

What changed between the healthy run and the problematic run?

Investigation

Compare Spark UI

Compare:

text

Old successful run vs New failed/slow run

Look at:

Input records

Input bytes

Shuffle read

Shuffle write

Peak memory

Spill memory

Spill disk

Task duration

Number of partitions

GC time

Data volume increased

Example:

text

Previously: 100 GB Now: 500 GB

The same job may no longer fit in the previous memory profile.

Data skew appeared

Example:

Previously:

text

Partition 1 → 2 GB Partition 2 → 2 GB Partition 3 → 2 GB

Now:

text

Partition 1 → 2 GB Partition 2 → 2 GB Partition 3 → 100 GB

One task spills heavily.

Query plan changed

Maybe:

text

Broadcast join

changed to:

text

Sort Merge Join

because the table became too large for the broadcast threshold.

That introduces a major shuffle/sort.

Partition count changed

Too few partitions can produce:

text

large partition → memory pressure → spill

Data type/schema changed

For example:

text

INT

became:

text

STRING

and causes larger memory usage or expensive operations.

Fix based on evidence

Possible fixes:

Repartition appropriately

Reduce shuffle

Filter/project earlier

Fix skew

Optimize joins

Avoid unnecessary sorts

Review broadcast strategy

Optimize file layout

Update Spark/Delta configuration where justified

Only after identifying the cause would I consider:

text

more executor memory

because adding memory can mask a data or query-design problem.
Pipeline has no monitoring and silently drops 2% of records for weeks. How do you find it and prevent recurrence?
I would start with data reconciliation.

Monitoring shouldn’t only answer:

“Did the job run?”

It must also answer:

“Did the expected data arrive correctly?”

Step 1 — Establish source-to-target counts

For every batch:

text

Source count ↓ Bronze count ↓ Silver count ↓ Gold count

Example:

text

Source = 10,000,000 Bronze = 10,000,000 Silver = 9,800,000 Gold = 9,800,000

Now I know the loss occurs between Bronze and Silver.

Step 2 — Add batch identifiers

Every record should ideally carry metadata such as:

batch_id

ingestion_timestamp

source_file

source_system

Then I can reconcile individual batches.

Step 3 — Compare business keys

Count unique keys:

sql

SELECT COUNT(DISTINCT employee_id) FROM source;

versus:

sql

SELECT COUNT(DISTINCT employee_id) FROM target;

Then identify missing records:

sql

SELECT employee_id FROM source EXCEPT SELECT employee_id FROM target;

Step 4 — Find where the records disappear

Common causes:

Bad join

Inner join removing unmatched rows

Incorrect filter

Null handling

Duplicate removal

Failed parsing

Schema mismatch

Late-arriving records

Bad incremental watermark

Step 5 — Build data-quality monitoring

Metrics:

input_count

output_count

rejected_count

duplicate_count

null_count

invalid_count

processing_time

Create thresholds:

text

Expected loss < 0.1% Actual loss = 2% → ALERT

Step 6 — Quarantine rejected records

Instead of silently dropping:

text

Valid records → Silver Invalid records → quarantine/error table

Store:

record

error_reason

batch_id

timestamp

pipeline_version

Then records can be replayed after correction.
How do you read and interpret a query execution plan? Optimize a query joining two 500M-row tables taking 40 minutes.
The key principle is:

Don’t optimize based on SQL appearance alone. Optimize based on the actual execution plan and runtime metrics.

Step 1 — Understand the logical plan

Suppose:

sql

SELECT ... FROM orders o JOIN customers c ON o.customer_id = c.customer_id WHERE o.order_date >= '2026-01-01';

First check:

Which filters are pushed down?

Which columns are projected?

Join condition

Aggregations

Sorts

Step 2 — Inspect physical plan

I look for:

text

Scan ↓ Filter ↓ Exchange ↓ Sort ↓ Join

Important operators:

Exchange

Usually indicates a shuffle.

Sort

Can consume substantial memory and spill to disk.

BroadcastHashJoin

Usually useful when one side is small enough.

SortMergeJoin

Common for large-to-large joins.

Filter

I check whether filtering occurs early.

Step 3 — Check actual runtime metrics

For two 500M-row tables, I would investigate:

Input size

Rows after filter

Shuffle bytes

Shuffle partitions

Task distribution

Spill

Skew

Join type

Step 4 — Reduce data before joining

Suppose only 50M of the 500M order rows are needed.

Don’t do:

text

500M JOIN 500M

and filter afterward.

Instead:

sql

WITH filtered_orders AS ( SELECT customer_id, order_date, amount FROM orders WHERE order_date >= '2026-01-01' ) SELECT ... FROM filtered_orders o JOIN customers c ON o.customer_id = c.customer_id;

Also select only required columns.

Instead of:

sql

SELECT *

use:

sql

SELECT customer_id, amount, order_date

This reduces I/O and shuffle volume.

Step 5 — Check partitioning

If both datasets are heavily shuffled:

text

500M ↓ Exchange ↓ Sort ↓ Join

I would assess whether partitioning can be improved.

Step 6 — Check skew

Run:

sql

SELECT customer_id, COUNT(*) FROM orders GROUP BY customer_id ORDER BY COUNT(*) DESC LIMIT 20;

If one customer has an enormous number of records, skew may create straggler tasks.

Potential solutions:

AQE skew join

Salting

Separate handling for hot keys

Repartitioning

Step 7 — Check broadcast possibility

If filtering reduces customers from:

text

500M

to:

text

50 MB

then a broadcast join could be appropriate.

But I would not broadcast a 500M-row table blindly.

Step 8 — Check storage/layout

For Delta:

File sizes

Small files

Partitioning

Data skipping

Statistics

Clustering/data layout

Table maintenance

can affect scan performance.

Step 9 — Validate the optimization

I would compare:

text

Before: 40 min Shuffle: 800 GB Spill: 300 GB After: ...

I would also validate:

row count

duplicate count

business results

because a faster query that changes the result is not a successful optimization.

Interview summary

“I would first inspect the physical execution plan and runtime metrics, identify whether the bottleneck is scan, shuffle, sort, skew, join strategy, or spill, then reduce data before the join, project only required columns, address skew, use broadcast only when justified, and optimize Delta layout where appropriate. Finally I’d compare runtime and data correctness against the baseline.”
How would you design incremental loading for a source with no reliable updated_at column?
This is a common real-world problem.

If updated_at cannot reliably identify changed records, I would not depend on it as the watermark.

I would first determine what incremental mechanism the source supports.

Option 1 — CDC

Best option if available.

Use:

text

INSERT UPDATE DELETE

events from:

Database transaction logs

CDC tools

Debezium

SQL Server CDC

Oracle redo/LogMiner-based mechanisms

Source-specific CDC

Then store:

primary_key

operation

change_sequence

change_timestamp

The sequence number is generally more reliable than application timestamps.

Option 2 — Increasing ID / sequence

If every new record gets:

text

id = 1001 id = 1002 id = 1003

I can use:

sql

WHERE id > last_processed_id

But this only captures new records.

It doesn’t detect updates to old records.

Option 3 — Hash-based change detection

If there is no CDC and no reliable timestamp, calculate a hash of business columns:

sql

hash( name, department, salary, location )

Example:

sql

SHA2( CONCAT_WS( '||', name, department, salary, location ), 256 )

Store:

employee_id

record_hash

On the next extraction:

text

Source hash ↓ Compare ↓ Same hash → unchanged Different → changed New key → insert Missing key → possible delete

Important problem with full hash comparison

If the source has 500M rows and we must extract all 500M rows every day, we’ve technically detected changes but haven’t solved the incremental extraction problem.

So I would ask whether the source supports:

CDC

transaction sequence

change tracking

audit table

triggers

source-side change log

partition-level modification metadata

Option 4 — Snapshot + hash

If the source only provides full snapshots:

text

Day 1 snapshot Day 2 snapshot

I can compare snapshots.

Conceptually:

text

Previous snapshot ↓ employee_id + hash ↓ Join ↑ Current snapshot

Classify:

text

new → INSERT hash diff → UPDATE missing → DELETE candidate same → unchanged

Option 5 — Source-side audit/change table

If I control the source database, I might request a change-log table:

text

employee_changes -------------------- employee_id operation changed_columns change_sequence changed_at

Then the data platform consumes that table incrementally.

Make the pipeline restartable

I would maintain a control table:

text

pipeline_control ----------------------- pipeline_name source_name last_successful_sequence last_run_time status

Only advance the watermark after the target write succeeds.

For example:

text

Read changes ↓ Transform ↓ Write Bronze ↓ Validate ↓ Write Silver ↓ SUCCESS ↓ Advance watermark

If the job fails before success:

text

watermark remains unchanged

so the next run can safely retry.

What I would say in an interview

“If there is no reliable updated_at, I first look for CDC, transaction sequence numbers, source audit tables, or change tracking. If none exists, I can use a full snapshot plus hash comparison, but that is not truly incremental extraction because the source still has to be scanned. For a scalable design, I’d prefer a source-side change mechanism. I’d maintain a durable control table containing the last successfully processed sequence and only advance it after the target transaction succeeds.”
What causes a Spark job to throw an OutOfMemoryError (OOM), and how do you troubleshoot it – driver vs. executor? How would you diagnose an executor OOM in the Spark UI – what do you look at first?

Spark OOM is one of the most common failures in offline jobs, but many teams troubleshoot by simply increasing memory. The right approach starts by quantifying memory consumption and distinguishing between Executor OOM, Driver OOM, and data skew scenarios.

First, quantify memory consumption

Use SizeEstimator to precisely estimate the actual memory footprint of an RDD, DataFrame, or any Java object:

python

import org.apache.spark.util.SizeEstimator val rdd = sc.parallelize(largeDataset) println(s"Estimated size: ${SizeEstimator.estimate(rdd)} bytes")

This is more accurate than the Web UI Storage page, which only shows serialized size.

Distinguish Executor OOM from Driver OOM

Executor OOM symptoms: java.lang.OutOfMemoryError: Java heap space with stack traces pointing to Shuffle, Sort, or aggregation operations. Root cause: Executor execution memory is insufficient to handle the current Task’s data volume. Common causes include too few shuffle partitions (one Task handles too much data), collect() or toLocalIterator() pulling large data into Executor memory, or UDFs caching oversized objects.

Driver OOM symptoms: Java heap space with stack traces pointing to DAGScheduler, TaskScheduler, or collect(). Root cause: Driver needs to collect all Executor results. Common causes include collect() pulling an entire DataFrame to the Driver, toLocalIterator() accumulating data on the Driver side, or broadcast variables exceeding spark.driver.maxResultSize.

Diagnosing Executor OOM in Spark UI — what to look at first

Open the Stages page in Spark Web UI and focus on two metrics: Shuffle Read data volume per Task and runtime. If most Tasks complete in seconds but a few Tasks take minutes or longer, and their Shuffle Read volume far exceeds others, data skew is causing hot spots.

Further check Shuffle Write and Shuffle Read statistics on the Stage details page. If Shuffle Spill (Disk) is large, Executor memory cannot hold shuffle data and it is spilling heavily to disk. Normally Shuffle Spill should be zero or near zero; if it reaches 50% or more of Shuffle Read volume, there is a clear memory bottleneck.

In the Executors tab, check the exit code of failed executors. Exit code 137 means the OS killed the container (SIGKILL), typically due to memory limit exceeded.

Memory parameter tuning

python

spark.executor.memory: 8g # Executor total heap memory spark.memory.fraction: 0.6 # Unified memory as proportion of heap (default 0.6) spark.memory.storageFraction: 0.5 # Storage memory as proportion of unified memory (default 0.5)

Unified memory = heap memory × memory.fraction, where storage memory = unified memory × memory.storageFraction, and execution memory = unified memory – storage memory.

Key tuning principles:

If the job is Shuffle/Aggregation-heavy (e.g., GROUP BY, JOIN), increase execution memory by lowering memory.storageFraction to 0.3-0.4.

If the job is cache-heavy (e.g., repeatedly querying the same DataFrame), increase storage memory by raising memory.storageFraction to 0.6-0.7.

When shuffle partitions are insufficient, increase spark.sql.shuffle.partitions (default 200) so each Task handles less data.

Special handling for data skew OOM

When one Task’s processing time is more than 10x that of other Tasks, data skew is present. Solutions include:

Salting: Add a random suffix to skewed keys, distribute to multiple Tasks, then aggregate again.

Broadcast Join: Broadcast the small table to all Executors, avoiding Shuffle.

Adaptive Execution: Spark 3.0+ AQE can automatically detect skewed partitions and split them: spark.sql.adaptive.skewJoin.enabled=true.

Parquet/Snappy read OOM

When reading Parquet files, OOM commonly occurs because compressed Blocks expand significantly after decompression in memory. Snappy decompression ratios are typically 3-5x. If a Parquet file’s Row Group Size is 256MB, it can reach 1GB after decompression. Solutions include reducing parquet.block.size at write time, increasing Executor memory or the number of Executors, and using spark.sql.parquet.filterPushdown=true to reduce the amount of data read.
Explain the Delta transaction log mechanics – what happens on every commit? What are .crc files, and what happens if one gets deleted manually?
Delta transaction log mechanics

At the core of Delta Lake’s transactional engine is the _delta_log directory — a highly structured append-only log that tracks every change to the table across versions, applications, and time. It acts as a Git-like commit history for your data.

Within _delta_log, you will typically find:

JSON files: One per commit, capturing metadata and file actions (e.g., 00000000000000000001.json, 00000000000000000002.json)

Checkpoint files: Parquet file summaries written every N commits to accelerate reads and recovery

CRC files: Integrity checks that ensure the log’s correctness/accuracy

What happens on every commit

Every time data is written to a Delta table — whether through a batch insert, an update, or a merge — Delta records the event by creating a new JSON commit file inside _delta_log. Each JSON file represents all changes made in one atomic transaction.

The log-centric design enables reproducibility, auditability, lineage, and complete table versioning without relying on external tooling. Delta Lake uses optimistic concurrency control: each job attempts to write the next versioned JSON file. If another job has already written it, the transaction retries using the next available number. Only one transaction can succeed at a given version.

Data files are written first, but only made visible after a successful JSON commit. If a job crashes midway, no partial state is exposed. Readers always interact with a consistent, completed version.

What are .crc files

A .crc file inside the _delta_log folder is a Cyclic Redundancy Check file — essentially a checksum used to verify the integrity of transaction log files. When Delta Lake writes a new JSON or checkpoint file to _delta_log, it also writes a corresponding .crc file containing a checksum hash of the data file. When the file is later read, the system compares the actual file contents to the checksum to ensure nothing was corrupted during storage or transfer.

CRC files play several critical roles:

Data Integrity: Detect corruption in transaction log files (e.g., due to disk failure or network issues).

Transaction Consistency: Ensure that only fully written, valid commits are considered part of the table’s history.

Efficient Validation: Allow quick verification without scanning the entire file.

Error Detection: When a file is read, its checksum is compared with the stored CRC value. If they do not match, it indicates corruption.

What happens if a .crc file gets deleted manually

CRC files are automatically managed by Delta Lake and Spark — you do not need to interact with them directly. During cleanup (e.g., after checkpointing), older .crc files may be deleted based on the delta.logRetentionDuration setting.

If a .crc file is deleted manually, Delta Lake can typically regenerate it from the corresponding transaction log file. The checksum is computed from the log file content, so as long as the actual log file (JSON or checkpoint Parquet) is intact, Delta can recalculate and recreate the .crc file. However, manual deletion is not recommended and can temporarily affect performance during validation. In practice, if a .crc file is missing, Delta will detect this and either recreate it or proceed without the optimization, but data integrity is not compromised as long as the underlying transaction log file is valid.

How would you set up a service principal for a GitHub Actions CI/CD pipeline deploying a Databricks Asset Bundle (DAB)? Walk through the structure of a databricks.yml file and how targets (dev/staging/prod) work. How would you parametrize the same DAB job to run against different catalogs per environment (dev_catalog vs. prod_catalog)?

Setting up a service principal for GitHub Actions

Step 1: Create the service principal in Databricks:

databricks service-principals create --json '{
  "display_name": "GitHub Actions CI",
  "active": true
}'

Note the application_id returned. Then create an OAuth secret:

databricks service-principal-secrets create \
  --service-principal-id <application_id>

Grant permissions to the service principal:

databricks permissions update workspace --json '{
  "access_control_list": [{
    "service_principal_name": "<application_id>",
    "permission_level": "CAN_MANAGE"
  }]
}'

Step 2: Configure GitHub secrets:

gh secret set DATABRICKS_HOST --body "https://adb-1234567890.1.azuredatabricks.net"
gh secret set DATABRICKS_CLIENT_ID --body "your-client-id"
gh secret set DATABRICKS_CLIENT_SECRET --body "your-client-secret"

For staging/prod environments, set additional secrets like DATABRICKS_HOST_STAGING, DATABRICKS_HOST_PROD, etc..

Structure of a databricks.yml file

The databricks.yml file defines the bundle configuration. Key sections include:

bundle:
  name: my-bundle

variables:
  catalog_name:
    description: The catalog to use
    default: dev_catalog

targets:
  dev:
    default: true
    workspace:
      host: ${DATABRICKS_HOST}
    variables:
      catalog_name: dev_catalog

  staging:
    workspace:
      host: ${DATABRICKS_HOST_STAGING}
    variables:
      catalog_name: staging_catalog

  prod:
    workspace:
      host: ${DATABRICKS_HOST_PROD}
    variables:
      catalog_name: prod_catalog

resources:
  jobs:
    my-job:
      name: my-job
      tasks:
        - task_key: my-task
          notebook_task:
            notebook_path: ./src/notebook.py
          existing_cluster_id: ${var.cluster_id}

How targets work

Targets define deployment environments. Each target can override workspace host, variables, and resource configurations. When you run databricks bundle deploy -t dev, it uses the dev target’s settings. The default: true flag marks a target as the default when no -t flag is provided.

Parametrizing the same DAB job for different catalogs

Use custom variables with the ${var.<variable_name>} syntax. Define a variable in the variables mapping, then reference it in resource definitions:

variables:
  catalog_name:
    description: The catalog for the environment
    default: dev_catalog

resources:
  jobs:
    my-job:
      name: my-job
      tasks:
        - task_key: my-task
          notebook_task:
            notebook_path: ./src/notebook.py
            base_parameters:
              catalog: ${var.catalog_name}

Then set the variable value per target:

targets:
  dev:
    default: true
    variables:
      catalog_name: dev_catalog

  staging:
    variables:
      catalog_name: staging_catalog

  prod:
    variables:
      catalog_name: prod_catalog

Alternatively, provide the value at deploy time:

databricks bundle deploy -t prod --var="catalog_name=prod_catalog"

You can also set environment variables with the prefix BUNDLE_VAR_ before calling bundle commands.

Auto Loader vs. COPY INTO – which scales better? Auto Loader typically needs separate paths per source; if that’s not allowed, what would you propose instead? Can a replay mechanism be built on top of Auto Loader? When would you use COPY INTO instead of Auto Loader?
Scaling comparison

Auto Loader scales better for large-scale ingestion. If you are expecting files in the order of millions or more over time, use Auto Loader. Auto Loader requires fewer total operations to discover files compared to COPY INTO and can split processing into multiple batches, which means Auto Loader is less expensive and more efficient at scale.

If you’re going to ingest files in the order of thousands over time, you can use COPY INTO. COPY INTO is idempotent and incremental but doesn’t scale as efficiently for millions of files.

When separate paths per source are not allowed

Auto Loader provides a cloudFiles Structured Streaming source that can process an input directory path. If you cannot use separate paths per source, you have several options:

Use glob patterns with a single root path: You can use glob patterns to filter directories, but be aware that glob patterns don’t limit Auto Loader’s initial file discovery scan. Auto Loader still evaluates all subdirectories under the specified root, and the glob pattern acts as a filter after the scan. This can impact performance, especially with partitions.

Use a more specific root path: Instead of using glob patterns, use a more specific root path to reduce the scope of the initial scan. For example, instead of /mnt/my_table/{year=2025/month=1/day=2,year=2025/month=1/day=3}, use /mnt/my_table/year=2025/month=1/{day=2,day=3}.

Use COPY INTO as a complement: You can use COPY INTO to reload a subset of files while an Auto Loader stream is running simultaneously. This is actually one of the advantages of COPY INTO over Auto Loader — loading a subset of re-uploaded files is easier to manage with COPY INTO.

Replay mechanism on Auto Loader

Auto Loader itself does not provide a built-in replay mechanism. With Auto Loader, it’s harder to reprocess a select subset of files. However, you can use COPY INTO to reload a subset of files while an Auto Loader stream is running simultaneously.

For more robust replay capabilities, Databricks Lakeflow pipelines (formerly Delta Live Tables) provide a “Rewind and replay” feature. Rewind restores table versions, streaming source offsets, and operator state together, so replay does not skip records or write duplicates. Rewind is for a recoverable pipeline that wrote incorrect data from a known point in time, such as after a bad transformation, malformed input, or a bad code deploy.

When to use COPY INTO instead of Auto Loader

Use COPY INTO when:

You are ingesting files in the order of thousands over time (not millions+).

You need to load a subset of re-uploaded files — this is easier to manage with COPY INTO than Auto Loader.

You want to reprocess a select subset of files while an Auto Loader stream is running.

Your data schema is stable and doesn’t evolve frequently (Auto Loader provides better primitives for schema inference and evolution).

You want to use SQL users and idempotent ingestion in Databricks SQL, notebooks, or Lakeflow Jobs.

What is Liquid Clustering, and how does it differ from Z-ORDER + partitioning? Can both be used on the same table?

What is Liquid Clustering

Liquid Clustering is a data layout optimization technique that replaces table partitioning and Z-ORDER. It is the recommended approach for organizing data in new Delta Lake tables. Unlike traditional partitioning, Liquid Clustering allows you to change your clustering strategy without rewriting existing data. This flexibility is valuable when query patterns evolve or when you’re uncertain which columns will be filtered most frequently.

To enable Liquid Clustering, use the CLUSTER BY clause when creating a table:

CREATE TABLE sales.transactions (
    transaction_id STRING,
    customer_id STRING,
    product_category STRING,
    region STRING,
    transaction_date DATE
)
CLUSTER BY (region, transaction_date);

You can specify up to four clustering keys. Choose columns that appear frequently in your query filters. The order of clustering keys matters less than selecting the right columns, as Liquid Clustering optimizes for all specified keys simultaneously.

Differences from Z-ORDER + partitioning

Aspect	Liquid Clustering	Z-Ordering	Partitioning
Key flexibility	Change keys without data rewrite	Changing keys requires full rewrite	Changing requires full rewrite
Write performance	Clustering on write for eligible operations	Applied only during OPTIMIZE	Applied at write time
Concurrent writes	Row-level concurrency supported	Transaction-level concurrency	Transaction-level concurrency
Compatibility	Requires Databricks Runtime 13.3 LTS or above	Supported in all runtimes	Supported in all runtimes

Liquid Clustering is incremental — OPTIMIZE only rewrites data as necessary to accommodate data that needs clustering. It does not rewrite data files with clustering keys that do not match the data being clustered.

Can both be used on the same table?

No, Liquid Clustering is not compatible with partitioning or Z-ORDER. For partitioned tables, evaluate whether the benefits of Liquid Clustering justify migrating to an unpartitioned table structure.

When migrating from Z-ORDER to Liquid Clustering:

Use the ZORDER BY columns as clustering keys
Run OPTIMIZE FULL to recluster all existing data

What is Change Data Feed (CDF), and how would you use it to build an incremental Silver-to-Gold pipeline?
What is Change Data Feed (CDF)

Change Data Feed is a Delta Lake feature that tracks row-level changes (inserts, updates, deletes) in the transaction log. When enabled, CDF generates additional files containing change data with metadata indicating the type of operation.

Enable CDF on a table:

sql

ALTER TABLE user_profiles SET TBLPROPERTIES (delta.enableChangeDataFeed = TRUE);

Query changes between versions:

sql

SELECT * FROM table_changes('user_profiles', 10, 20);

The transaction log records CDF files using the cdc action. Each CDF record includes:

The changed row data

_change_type: “insert”, “update_preimage”, “update_postimage”, or “delete”

_commit_version: Transaction log version where the change occurred

_commit_timestamp: Timestamp of the change

Building an incremental Silver-to-Gold pipeline using CDF

In a medallion architecture, you can use CDF to propagate changes from Silver to Gold incrementally:

Enable CDF on the Silver table: Because you’ll propagate MERGE operations downstream to the Gold layer, you need to enable Delta Lake CDF:

sql

ALTER TABLE silver_table SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

Capture changes from Silver using table_changes: Use the table_changes function to select all table modifications from a specific Delta version to another:

sql

SELECT * FROM table_changes('silver_table', 10, 20);

Process CDF output: The CDF stream contains four change types:

update_preimage: Content of the row before an update

update_postimage: Content of the row after the update (what you want to capture downstream)

delete: Content of a row that has been deleted

insert: Content of a new row that has been inserted

One update results in two rows in the CDC stream (one with previous values, one with new values).

Apply changes to Gold: For a Gold layer that should reflect Silver changes, you need to propagate deletes and updates. You can use the CDF data to perform MERGE operations on the Gold table, applying the same modifications.

Deduplication: Since you’re consuming CDF data, you need to perform deduplication. Multiple changes to the same key may appear in the CDF stream, and you need to ensure the final state is correct.

CDF is essential for CDC pipelines, incremental processing (processing only changed data rather than full table scans), audit trails, and real-time analytics.
Explain watermarking in Structured Streaming and why it’s needed for stateful aggregations.
What is watermarking

A watermark is a threshold that defines how late data is allowed to arrive in a streaming query. You define a watermark using withWatermark("eventTime", delay) on a streaming DataFrame, where eventTime is the column containing the event timestamp and delay is the maximum lateness you’re willing to tolerate.

For example:

python

streamingDf.withWatermark("event_time", "10 minutes")

Why watermarking is needed for stateful aggregations

Stateful operations in Structured Streaming — such as windowed aggregations, stream-stream joins, and deduplication — maintain state information in memory. Without a mechanism to expire old state, this state grows indefinitely and can cause memory issues.

Watermarks solve this by allowing Spark to discard state information for data that is considered “too late” based on the watermark threshold. This bounds the total amount of state information in memory.

How watermarks work with windowed aggregations

In a groupBy() and window() operation, state information is maintained for each window until the end of the window is older than the watermark (the latest observed event time minus the lateness threshold). At that point, the aggregation state is dropped.

Output modes control when a query with watermarks writes data to the sink:

Append mode: The query writes rows to the target table after the watermark threshold has passed. All writes are delayed based on the lateness threshold. Old aggregation state is dropped after the threshold has passed.

Update mode: The query writes rows to the target table as results are calculated, and the query can update and overwrite rows as new data arrives.

Complete mode: Aggregation state isn’t dropped. The query rewrites the target table for each trigger.

Tradeoffs in watermark duration

Selecting a watermark duration has tradeoffs:

Shorter watermarks lower query latency because queries store less state information and write results after each watermark duration completes. However, short watermarks have low tolerance for late data.

Longer watermarks have high tolerance for late data. However, long watermarks increase query latency because queries must store more state information and wait to write results after a longer watermark duration.

Watermarks for stream-stream joins

For inner joins, Databricks recommends that you set a watermark threshold on each streaming data source to allow the query to discard state information for old records. Without watermarks, Structured Streaming attempts to join every key from both sides of the join on each trigger, which might affect performance. For outer joins, watermarking is mandatory.
What happens when you execute a query in Databricks? What is Adaptive Query Execution (AQE), and what 3 things does it do at runtime?
What happens when you execute a query

When you execute a query in Databricks, the query goes through several stages:

Parsing and analysis: The SQL or DataFrame operations are parsed into a logical plan.

Optimization: The logical plan is optimized using Catalyst optimizer (predicate pushdown, column pruning, etc.).

Physical planning: The optimized logical plan is converted into a physical plan with specific join strategies, partitioning schemes, etc.

Execution: The physical plan is executed as a series of stages with tasks distributed across executors.

Adaptive Query Execution (AQE) is a runtime query re-optimization feature that changes this flow by re-optimizing the query plan during execution, after shuffle and broadcast exchanges complete, when Spark has the most up-to-date and accurate statistics.

What AQE does at runtime

AQE has four major features (though the question asks for three — these are the primary ones):

Dynamically changes sort merge join into broadcast hash join: If runtime statistics show one side of the join is small enough, AQE converts a SortMergeJoin (which requires shuffle) into a BroadcastHashJoin (which avoids shuffling the large side).

Dynamically coalesces partitions: After a shuffle exchange, AQE combines small partitions into reasonably sized partitions. Very small tasks have worse I/O throughput and suffer more from scheduling and task setup overhead. Combining small tasks saves resources and improves cluster throughput.

Dynamically handles skew in sort merge join and shuffle hash join: AQE detects skewed partitions at runtime and splits (and replicates if needed) skewed tasks into roughly evenly sized tasks, preventing stragglers from dominating execution time.

Dynamically detects and propagates empty relations: AQE can detect when a relation is empty at runtime and propagate that information through the plan, potentially eliminating unnecessary work.

How to observe AQE in query plans

AQE-applied queries contain one or more AdaptiveSparkPlan nodes, usually as the root node of each main query or sub-query. Before the query runs or while it is running, the isFinalPlan flag shows as false; after the query execution completes, it changes to true.

The query plan diagram evolves as execution progresses and reflects the most current plan. Nodes that have already been executed (with metrics available) will not change, but those that haven’t can change over time as the result of re-optimizations.
Explain Deletion Vectors — what problem do they solve?
What are Deletion Vectors

Deletion Vectors are a storage optimization feature in Delta Lake. By default, when a single row in a data file is deleted, the entire Parquet file containing that record must be rewritten. With Deletion Vectors enabled for the table, DELETE, UPDATE, and MERGE operations use deletion vectors to mark existing rows as removed or changed without rewriting the Parquet file. Subsequent reads on the table resolve the current table state by applying the deletions indicated by deletion vectors to the most recent table version.

What problem they solve

Deletion Vectors solve the problem of expensive row-level modifications in Delta Lake. In traditional Delta Lake (without Deletion Vectors):

DELETE: Even deleting a single row requires rewriting the entire Parquet file containing that row.

UPDATE/MERGE: Updating a single row similarly requires rewriting the entire file.

This is inefficient for workloads with frequent small updates or deletes. Deletion Vectors allow these operations to mark rows as logically deleted or changed without physically rewriting the file, making them much faster and more efficient.

How to enable Deletion Vectors

Deletion Vectors can be enabled at table creation or via ALTER:

sql

CREATE TABLE <table-name> TBLPROPERTIES ('delta.enableDeletionVectors' = true); ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.enableDeletionVectors' = true);

When changes are physically applied

Deletion vectors indicate changes to rows as soft-deletes. These changes are applied physically when one of the following events causes the data files to be rewritten:

An OPTIMIZE command is run on the table

Auto-compaction triggers a rewrite of a data file with a deletion vector

REORG TABLE ... APPLY (PURGE) is run against the table

Compatibility

When you enable Deletion Vectors, the table protocol is upgraded. After upgrading, the table will not be readable by Delta Lake clients that do not support Deletion Vectors. Databricks Runtime 12.2 LTS and above supports reading tables with Deletion Vectors enabled. Databricks Runtime 14.2 and above supports row-level concurrency for tables with Deletion Vectors.

Databricks recommends using Databricks Runtime 14.3 LTS and above to write tables with Deletion Vectors to use all optimizations.
Explain how Databricks lineage gets created in Unity Catalog — how would you use it to assess the blast radius of a schema change?

Unity Catalog captures data lineage automatically at runtime for queries that run on Databricks, down to the column level, and aggregates it across all workspaces attached to the metastore. No additional configuration or code changes are required — as long as tables are registered in Unity Catalog and queries use Spark DataFrame or Databricks SQL interfaces.

How lineage is created

The capture happens on Unity Catalog-enabled compute, including shared, single-user, or serverless clusters, and SQL warehouses with Unity Catalog enabled. When a query reads from or writes to a Unity Catalog table, the compute automatically records the relationship. Clusters that are not Unity Catalog-enabled, or queries that access tables by direct file path instead of the three-level namespace (catalog.schema.table), do not generate lineage data.

Lineage data includes the tables and views involved, the notebooks, jobs, and dashboards that interact with them, and column-level dependencies showing how individual columns derive from source columns. The lineage graph is aggregated across all workspaces sharing the same metastore, so lineage captured in one workspace is visible in others.

Using lineage to assess blast radius of a schema change

Before changing or deleting a table or column, you can use lineage to identify all downstream dependencies. The workflow is:
1. Open Catalog Explorer and find the table you plan to modify.
2. Select the Lineage tab to see related tables in the lineage panel, then click See Lineage Graph for an interactive visualization.
3. Explore downstream nodes — by default, one level is displayed; click the plus icon on any node to reveal more connections.
4. Filter by asset type — use the Lineage details panel to see which notebooks, jobs, pipelines, and dashboards depend on the table.
5. Check column-level lineage — click a column in the graph to see which downstream columns derive from it. This is critical for assessing the impact of dropping or renaming a specific column.
For programmatic analysis, you can query the lineage system tables directly:
- system.access.table_lineage records read and write events at the table level
- system.access.column_lineage tracks column-level dependencies
A query to find downstream tables would look like:
sql
```
SELECT target_table_full_name
FROM system.access.table_lineage
WHERE source_table_full_name = 'catalog.schema.table_name'
```
Practical considerations

Lineage data retention varies by access method. Catalog Explorer and the lineage API retain data indefinitely for events after September 1, 2024. Lineage system tables retain a rolling one-year window. For compliance audits or historical impact analysis beyond one year, use Catalog Explorer or the API.
What is Databricks Lakeflow / Delta Live Tables (DLT), and when would you choose it over hand-written PySpark jobs?

Databricks Lakeflow is an end-to-end data engineering solution that unifies ingestion, transformation, and orchestration. It includes Lakeflow Connect (ingestion connectors), Lakeflow pipelines (formerly Delta Live Tables), Lakeflow Designer (visual data prep), and Lakeflow Jobs (orchestration).

Lakeflow pipelines, built on Apache Spark Declarative Pipelines (SDP), provide a declarative framework for building batch and streaming pipelines in SQL and Python. You define what the pipeline should produce — streaming tables, materialized views, and sinks — and the engine handles execution order, retries, and scaling automatically.

What Lakeflow pipelines provide
- Automatic orchestration: The pipeline analyzes relationships between flows and determines the correct execution order, maximizing parallelism without manual orchestration code.
- Built-in data quality: Declarative expectations validate data during processing; the system tracks violations and can halt processing if data quality is compromised.
- Incremental processing: An incremental engine keeps materialized views up to date, reprocessing only new or changed source data when possible.
- Unified batch and streaming: The same pipeline definition handles both batch and streaming workloads.
- Operational visibility: Lineage tracking, execution graphs, and monitoring dashboards are included without additional configuration.
- Auto CDC API: Handles SCD Type 1 and Type 2 change data capture events, including out-of-order events, without manual watermarking code.
When to choose Lakeflow pipelines over hand-written PySpark

Choose Lakeflow pipelines when:
- Standard ETL patterns dominate: Ingestion from cloud storage, schema evolution, slowly changing dimensions — these reduce thousands of lines of code to a few declarative statements.
- Production pipelines with repeatable patterns: The declarative model reduces operational overhead and improves maintainability.
- Data quality must be enforced systematically: Expectations are defined in the pipeline itself, not scattered across validation code.
- Dependency management is complex: The engine handles execution order automatically, eliminating manual orchestration.
- Operational visibility is required: Lineage and monitoring come built-in.
- External streaming sinks are needed: SDP supports Kafka, Event Hubs, Delta tables, and custom Python sinks.
Choose hand-written PySpark when:
- Complex procedural logic is required that is difficult to express declaratively.
- One-time analysis or exploratory work rather than recurring production pipelines.
- Custom integrations with external systems, ML feature engineering, or API calls are central to the workload.
- Full control over execution is needed — partitioning, broadcast joins, caching strategies.
A hybrid approach is common: ingestion in declarative pipelines, enrichment in notebooks, orchestration in Lakeflow Jobs.

Explain broadcast join vs. sort-merge join vs. shuffle-hash join. When does Spark choose each?

Spark uses three primary physical join strategies for equi-joins:

Broadcast Hash Join (BHJ)

The small table is collected to the Driver, serialized, and broadcast to every Executor. Each Executor builds an in-memory hash table from the small table and streams the large side through locally, probing the hash table per row. No shuffle of the large side occurs.

Spark chooses BHJ automatically when one side’s estimated size is below spark.sql.autoBroadcastJoinThreshold, default 10 MB. You can force it with the broadcast() hint or /*+ BROADCAST(t) */ SQL hint. This is the cheapest join and the biggest performance lever when applicable.

Caution: broadcasting too large a table blows up the Driver (which collects the small side first) and every Executor’s memory. Broadcast is for single-digit to low-tens of MB tables.

Sort-Merge Join (SMJ)

Both sides are shuffled by the join key, sorted by key, then merged partition by partition. This is the default strategy for large-to-large joins when spark.sql.join.preferSortMergeJoin=true (the default). It scales to any size but pays for two shuffles and two sorts. Recognize it in execution plans by the SortMergeJoin node with Exchange hashpartitioning on both sides.

Shuffle Hash Join (SHJ)

Both sides are shuffled by key, but the smaller shuffled side builds a hash table instead of sorting. This avoids the sort but requires the smaller side’s partition to fit in memory. It is rarely chosen automatically because the default configuration prefers sort-merge; it typically requires an explicit hint or disabling preferSortMergeJoin.

Decision guide

Situation	Strategy	Shuffle?	Plan node
One side < ~10 MB (dimension table)	Broadcast Hash Join	No (large side stays put)	BroadcastHashJoin
Both sides large	Sort-Merge Join	Yes, both sides + sort	SortMergeJoin
Both large, one fits in memory after shuffle, no sort wanted	Shuffle Hash Join	Yes, both sides	ShuffledHashJoin

The instinct to build: fact joined to dimension → broadcast the dimension. Fact joined to fact → sort-merge or shuffle-hash depending on size.

Explain persist() vs. cache(), and the different StorageLevels. How does Spark’s Unified Memory Manager handle the conflict when execution memory needs space that’s currently occupied by cached (storage memory) data — what actually gets evicted, and when?

persist() vs. cache()

cache() is a shorthand for persist(StorageLevel.MEMORY_AND_DISK_DESER) in recent Databricks Runtime versions. persist() allows you to specify exactly how data is stored.

The core value of manual caching is avoiding repeated expensive computations — not just avoiding disk reads. If an RDD or DataFrame is built by reading 100 GB from a database and running 10 complex transformations, reusing a cached version saves re-executing all that work.

StorageLevels

MEMORY_ONLY: Default for persist() in classic Spark. Objects stored deserialized in JVM memory. Most CPU-efficient. If the RDD doesn’t fit, partitions are recomputed on demand.

MEMORY_ONLY_SER: Serialized objects in memory. More space-efficient but slower to access. Use with a fast serialization library.

MEMORY_AND_DISK: Spills partitions that don’t fit in memory to disk.

MEMORY_AND_DISK_SER: Serialized with disk spillover.

DISK_ONLY: All data on disk.

OFF_HEAP: Stores data outside the JVM heap to reduce GC pressure and allow multiple Executors to share memory.

Replicated variants (e.g., MEMORY_ONLY_2): Keep two replicas for faster fault recovery without recomputation.

Unified Memory Manager conflict resolution

Spark’s Unified Memory Manager (UMM) enforces a soft boundary between execution and storage memory. Both can borrow from each other, but with different priorities.

The shared region is (total heap - 300 MB) × spark.memory.fraction (default 0.6 or 0.75 depending on version). Within this, spark.memory.storageFraction (default 0.5) determines the initial storage region boundary.

When execution memory needs space occupied by cached data:

The UMM’s acquireExecutionMemory method handles this. If execution memory is insufficient, it attempts to reclaim memory from storage:

First, reclaim free storage memory: Any unused portion of the storage pool is immediately usable.

Second, evict cached blocks: If storage has grown beyond its initial region (borrowed from execution), execution can evict cached blocks until sufficient memory is reclaimed. Eviction follows least-recently-used (LRU) order.

Reclaim up to the initial storage region size: The storageRegionSize is the floor — execution cannot evict storage below this boundary.

Critical asymmetry: Execution memory is never evicted by storage. If execution has consumed most of the memory, attempts to cache new blocks will fail — new blocks are evicted immediately according to their storage level. This means caching can become ineffective under heavy execution memory pressure.

Practical implication: In a job that caches large intermediate results and also runs complex multi-stage aggregations, the aggregation (execution) can evict the cache. The cache hit rate drops, causing more disk I/O and potentially worse performance than not caching at all. This is why spark.memory.storageFraction should be tuned based on workload type: lower it (0.3-0.4) for shuffle/aggregation-heavy jobs, raise it (0.6-0.7) for cache-heavy iterative workloads.

Deep Clone vs. Shallow Clone – note that renaming a table also renames the underlying storage directory.

Deep Clone

Deep Clone copies both data and metadata from the source table to the clone target. The clone is fully independent — changes to the clone do not affect the source, and vice versa. For Delta tables, stream and COPY INTO metadata are also cloned, allowing streaming workloads to be stopped on the source and resumed on the clone from where they left off.

Deep Clone is more expensive because it copies all data files. It is the right choice when you need a standalone copy for testing, migration, or when the source table may undergo VACUUM operations that would break a shallow clone.

Shallow Clone

Shallow Clone copies only metadata — schema, partitioning information, invariants, nullability, and TBLPROPERTIES — and references the source table’s data files directly. No data files are copied, making it cheaper in both compute and storage.

The critical dependency: because data files remain in the source location, running VACUUM on the source table can make the shallow clone unreadable with a FileNotFoundException. In that case, running CLONE with replace on the shallow clone repairs it.

What is not cloned

Table description, user-defined commit metadata, Delta Lake table history, and Unity Catalog properties such as tags are not cloned. The cloned table has its own independent history — time travel queries on the clone do not work with the same inputs as the source.

Renaming and storage directories

When you rename a table, the underlying storage directory is also renamed. This matters for shallow clones: if you shallow-clone a table and then rename the source table, the shallow clone’s reference to the original storage directory breaks. The clone points to the old directory path, which no longer exists. This is another reason shallow clones are fragile and should be used only when the source table’s location is stable and not subject to VACUUM or rename operations.

When to use each

Scenario	Choice
Standalone copy for testing or migration	Deep Clone
Source table may be VACUUMed or renamed	Deep Clone
Quick metadata-only copy for development	Shallow Clone
Source location is stable and not VACUUMed	Shallow Clone
Need stream/COPY INTO metadata	Deep Clone (only)

Explain clustered vs. non-clustered indexing, and how a wrong index choice can make a query slower, not faster.
Clustered Index

A clustered index defines the physical order of data rows in the table. Because rows can be stored in only one order, there can be only one clustered index per table. In InnoDB (MySQL) and SQL Server, the primary key is typically the clustered index.

Clustered indexes are optimal for:

Range queries: Since data is stored in key order, range scans read contiguous disk pages.

Sort operations: Data is already in the desired order, eliminating sort cost.

Sequential access patterns: Records accessed sequentially benefit from physical locality.

The clustered index key should be narrow, unique or highly selective, and stable.

Non-Clustered Index

A non-clustered index is a separate structure containing the indexed key values plus a pointer to the corresponding data row. You can have many non-clustered indexes per table.

Non-clustered indexes are ideal for:

Specific predicate lookups: Fast lookups for values not aligned with the clustered key.

Join operations: Indexing foreign key columns for efficient join lookups.

Sorting patterns: Supporting ORDER BY clauses not covered by the clustered index.

Covering queries: Including extra non-key columns to avoid key lookups.

How a wrong index choice makes queries slower

Low selectivity columns: Indexing a boolean is_active column with only two values (true/false) means the index points to roughly half the table. The optimizer may still choose a full table scan, and the index adds maintenance overhead without query benefit.

Large result sets: When a query returns a large percentage of table rows, a sequential scan is often faster than thousands of random index lookups. The optimizer may correctly ignore the index, but the index still slows down writes.

Non-covering non-clustered index: If a query needs columns not included in the non-clustered index, each index seek triggers a bookmark lookup (key lookup) to fetch the full row from the clustered index or heap. This random I/O can make the query slower than a simple table scan.

Write amplification: Every INSERT, UPDATE, or DELETE must maintain all indexes on the table. A table with many indexes has slower write throughput. If indexes are not being used by queries, they are pure overhead.

Clustered index on a poor key: A clustered index on a wide, frequently changing column (e.g., a string identifier that changes) causes page splits and fragmentation. Rows physically reorder on updates, degrading performance over time.

Wrong clustered index for the workload: If most queries filter by created_at but the clustered index is on id, range scans on created_at must follow non-contiguous pointers through the clustered index or use a non-clustered index with lookups.

The key principle: indexes help queries that filter, sort, or join on indexed columns. A query that reads every row (like SELECT COUNT(*)) won’t benefit from a non-covering index. Over-indexing or mis-indexing creates write overhead and can degrade read performance when the optimizer chooses a suboptimal path.

Explain a Databricks Service Principal – how do you create one, and what’s it used for vs. a normal user identity?

What is a Service Principal

A service principal is a specialized identity in Databricks designed for automation and programmatic access. It provides secure, API-only access to Databricks resources for automated tools, scripts, and CI/CD platforms, without relying on individual user credentials.

Service principals give automated tools and scripts API-only access to Databricks resources, providing greater security than using user accounts. You can grant and restrict a service principal’s access to resources in the same way as a Databricks user — grant account admin or workspace admin roles, grant data access via Unity Catalog, or add it as a member to a group.

How to create a Service Principal

Using the Databricks CLI:

databricks service-principals create \
  --display-name "GitHub Actions CI"

Note the applicationId and id values from the output.

Generate an OAuth secret:

databricks service-principal-secrets create \
  --service-principal-id <id>

Copy the secret value from the output.

Configure authentication:

export DATABRICKS_HOST=<workspace-url>
export DATABRICKS_CLIENT_ID=<application-id>
export DATABRICKS_CLIENT_SECRET=<oauth-secret>

Use cases vs. normal user identity

Aspect	Service Principal	Normal User Identity
Purpose	Automation, CI/CD, scheduled jobs	Interactive work, exploration
Credentials	API-only, OAuth secrets or PATs	Personal credentials, SSO
Lifecycle	Independent of any employee; can be disabled or deleted without impacting users	Tied to employee; leaving organization breaks jobs
Security	Restricted access independently of users; no interactive login	Subject to user permissions and SSO policies
Job ownership	Jobs run as service principal; survive employee departure	Jobs owned by user fail if user is deactivated

Common use cases

CI/CD pipelines: Deploy notebooks, libraries, and configurations automatically as part of continuous integration and deployment workflows.
Scheduled jobs: Run ETL pipelines, data processing jobs, and automated reports on a schedule without depending on individual user accounts.
Cross-system integrations: Connect external applications and services to Databricks for data ingestion, transformation, or analytics.
Automated testing: Execute integration tests and validate data pipelines as part of a testing framework.
Infrastructure as code: Provision and manage Databricks resources using Terraform, ARM templates, or Declarative Automation Bundles.

Why use a service principal instead of a user token

For CI/CD and automation, Databricks recommends using a service principal and its token instead of a user’s personal access token. Benefits include:

You can grant and restrict access independently of any user. For example, prohibit the service principal from acting as workspace admin while allowing specific users to remain admins.
Users can protect their personal access tokens from being accessed by CI/CD platforms.
You can temporarily disable or permanently delete a service principal without impacting other users — useful if the principal is suspected of being compromised.
If an employee leaves the organization, you can remove that user without affecting any service principals.

What is Salting and how does it help with Data Skew?

The Problem: Data Skew

Data skew happens when data is unevenly distributed across partitions. For example, if you join or groupBy on a key like country, and 60% of records have country = "USA", then one partition gets 60% of the data while others get tiny fractions. Result: one task runs for hours while 199 others finish in seconds. This is the classic “straggler task” problem.

What is Salting?

Salting = artificially adding a random suffix (a “salt”) to skewed keys to spread them across multiple partitions, then removing the effect after computation.

How It Works

Example: Joining a skewed transactions table with a small customers table on customer_id, where 1 customer has 10M records.

Step 1 — Add salt to the skewed (large) DataFrame:
python
```
from pyspark.sql.functions import rand, floor, concat, lit, col

salt_range = 10  # number of buckets

skewed_df = transactions.withColumn(
    "salted_key",
    concat(col("customer_id"), lit("_"), floor(rand() * salt_range).cast("int"))
)

# Duplicate the small DataFrame for each salt value
from functools import reduce
small_dfs = [
    customers.withColumn("salted_key", concat(col("customer_id"), lit("_"), lit(i)))
    for i in range(salt_range)
]
salted_small = reduce(lambda a, b: a.union(b), small_dfs)

# Now join on salted_key — skewed customer is split into 10 partitions
result = skewed_df.join(salted_small, on="salted_key")
```
Step 2 — Aggregate back (salt is no longer needed):
python
```
result.groupBy("customer_id").agg(...)
```
Why It Helps

Without Salting With Salting

1 partition gets 10M rows 10 partitions get ~1M rows each

1 task = 3 hours All tasks = ~18 min each

Cluster underutilized Full parallelism

When to Use Salting
- Large-to-large joins with a few hot keys
- groupBy aggregations where a few keys dominate
- Not needed for broadcast joins (small table) — that’s a different fix.
Alternative Fixes for Skew
1. Broadcast join — if one side is small (< 10MB default).
2. Adaptive Query Execution (AQE) in Spark 3.x — spark.sql.adaptive.skewJoin.enabled = true handles it automatically.
3. Isolate + process hot keys separately, then union results.
4. Pre-aggregate before join.
Explain Bloom Filters and how they improve Join Performance

What is a Bloom Filter?

A probabilistic data structure that answers: “Is this element in the set?”
- Returns “definitely not present” OR “possibly present”
- No false negatives, but possible false positives
- Very memory-efficient: ~10 bits per element for 1% false positive rate
How It Works Internally
1. A bit array of size m (all zeros initially).
2. k independent hash functions.
3. Insert(x): hash x with all k functions → set those k bit positions to 1.
4. Query(x): hash x with all k functions → if any bit is 0 → definitely NOT in set. If all are 1 → possibly in set.
How It Improves Join Performance

Scenario: Joining fact_sales (1 TB) with dim_product (10 GB) on product_id.

Without Bloom Filter:
- Spark shuffles ALL 1 TB of fact data across the network.
- Most product_ids in fact don’t even exist in dim → wasted shuffle.
With Bloom Filter (Dynamic Partition Pruning / DPP in Spark 3.x):
sql
```
-- Spark automatically builds a bloom filter on dim_product.product_id
-- and pushes it down to filter fact_sales BEFORE the shuffle
SELECT * FROM fact_sales f
JOIN dim_product d ON f.product_id = d.product_id
```
Spark:
1. Builds a Bloom filter from the smaller table’s join keys.
2. Broadcasts the filter.
3. Filters the larger table before shuffling — eliminating rows that can’t match.
4. Only the reduced dataset is shuffled.
Result: 60–90% reduction in shuffle I/O → 3–5x faster joins.

Enabling It in Spark
python
```
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.useStats", "true")
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio", "0.5")
```
Other Uses of Bloom Filters
- HBase / Cassandra / RocksDB — avoid disk lookups for missing keys.
- Snowflake / Delta Lake — file-level pruning (skip files that can’t contain a key).
- Spark’s bloomFilter aggregate function for manual implementation.
Trade-offs

Pro Con

Tiny memory footprint False positives (rare)

O(k) insert/query Can’t delete elements

Massive I/O savings Tunable accuracy vs. size

How does Spark Memory Management work — Heap, Caching, GC?

Spark Memory Model (Unified Memory Management, Spark 1.6+)

Each Executor JVM has:

┌─────────────────────────────────────────────┐
│            JVM Heap (spark.executor.memory) │
│  ┌───────────────────────────────────────┐  │
│  │   Reserved Memory (300 MB fixed)      │  │
│  ├───────────────────────────────────────┤  │
│  │   User Memory (40% of usable)         │  │  ← Your data structures, UDFs
│  ├───────────────────────────────────────┤  │
│  │   Spark Memory (60% of usable)        │  │
│  │  ┌─────────────────┬───────────────┐  │  │
│  │  │ Execution (50%) │ Storage (50%) │  │  │  ← Unified, can borrow
│  │  │ joins, sorts,   │ cache,        │  │  │
│  │  │ aggregations,   │ broadcast,    │  │  │
│  │  │ shuffles        │ persist       │  │  │
│  │  └─────────────────┴───────────────┘  │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘
       + Off-heap (spark.memory.offHeap)

Key Configs

Config	Default	Purpose
`spark.executor.memory`	1g	Total JVM heap
`spark.memory.fraction`	0.6	Fraction of heap for Spark memory
`spark.memory.storageFraction`	0.5	Fraction of Spark memory for caching
`spark.memory.offHeap.enabled`	false	Use off-heap memory
`spark.executor.memoryOverhead`	10%	Off-heap (JVM overhead, Python, etc.)

Unified Memory: Execution ↔ Storage

Execution memory (shuffles, joins, sorts) can evict cached data if needed.
Storage memory (cache/persist) cannot evict execution memory.
This prevents OOM during heavy joins while still allowing caching.

Caching Strategies

from pyspark import StorageLevel

df.cache()                          # MEMORY_AND_DISK (default)
df.persist(StorageLevel.MEMORY_ONLY)          # Fast, but may recompute
df.persist(StorageLevel.MEMORY_AND_DISK)      # Spills to disk
df.persist(StorageLevel.DISK_ONLY)            # Slow but memory-safe
df.persist(StorageLevel.MEMORY_ONLY_SER)      # Serialized — less memory, more CPU
df.persist(StorageLevel.OFF_HEAP)             # Off-heap (Tungsten)

Rule of thumb: Use MEMORY_AND_DISK_SER when memory is tight; MEMORY_ONLY when DF is reused many times and fits in RAM.

Garbage Collection (GC)

Spark executors run on the JVM → GC pauses hurt performance.

Symptoms: Long GC pauses in Spark UI (“GC Time” column), “Task not serializable”, OOM.

Tuning:

G1GC (default in Spark 2.x+) — low pause, concurrent.

--conf spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35"

Reduce object creation: Use DataFrames/Datasets (Tungsten) instead of RDDs.

Kryo serialization (faster, smaller than Java serialization):

spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

Avoid large collect() on driver.
Increase heap if you see frequent full GCs.

Common Memory Errors

Error	Cause	Fix
`OutOfMemoryError: Java heap space`	Executor heap too small	Increase `spark.executor.memory`
`Container killed by YARN`	Memory overhead exceeded	Increase `spark.executor.memoryOverhead`
`GC overhead limit exceeded`	Too much GC, little work	Increase heap or reduce caching
Driver OOM	`collect()`, large broadcast	Increase `spark.driver.memory`

Difference between SparkContext and SparkSession

Aspect	SparkContext (sc)	SparkSession (spark)
Introduced	Spark 1.0	Spark 2.0
Role	Gateway to Spark cluster; core entry point	Unified entry point for SQL, DataFrame, Dataset APIs
Package	`pyspark.SparkContext`	`pyspark.sql.SparkSession`
Purpose	RDDs, accumulators, broadcast, job scheduling	DataFrames, SQL, catalog, UDFs, config
Relationship	Wraps SparkContext internally	Contains a SparkContext (`spark.sparkContext`)
RDD Support	Primary API	Accessible via `spark.sparkContext`
SQL Support	None	`spark.sql()`, `spark.catalog`
Creation	`sc = SparkContext(conf)`	`spark = SparkSession.builder.appName("x").getOrCreate()`
Status	Still exists but rarely used directly	Recommended entry point

Code Comparison

Spark 1.x (old):

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext

conf = SparkConf().setAppName("app").setMaster("local[*]")
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
df = sqlContext.createDataFrame(data, schema)

Spark 2.x+ (recommended):

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("app") \
    .master("local[*]") \
    .config("spark.sql.shuffle.partitions", "200") \
    .enableHiveSupport() \
    .getOrCreate()

sc = spark.sparkContext  # still accessible if needed
df = spark.createDataFrame(data, schema)
spark.sql("SELECT * FROM table")

Key Takeaway

SparkSession = superset of SparkContext + SQLContext + HiveContext.
In Spark 2.x+, always use SparkSession; it’s the single entry point.
In Databricks/Snowpark/Glue, spark is auto-created — no need to build it.

Explain Repartition vs Coalesce with Use Cases

Quick Comparison

Feature	repartition()	coalesce()
Shuffle?	✅ Yes (full shuffle)	❌ No (unless increasing partitions)
Partition count	Can increase OR decrease	Can only decrease (without shuffle)
Data distribution	Even (uses hash)	Uneven (merges adjacent partitions)
Cost	Expensive (network I/O)	Cheap (just merges)
Use case	Fix skew, increase parallelism, partition by column	Reduce partitions before write, avoid small files
Guarantees	Balanced partitions	Faster but may be skewed

`repartition()`

df.repartition(200)                        # Even distribution, full shuffle
df.repartition(200, "country")             # Hash partition by column
df.repartition("country", "date")          # Multiple columns

When to use:

Skewed partitions → rebalance data.
Increasing parallelism → more partitions = more tasks.
Before a join → co-partition both DFs by join key to avoid shuffle:
python
```
df1.repartition(200, "id").join(df2.repartition(200, "id"), "id")
```
Before writing partitioned data → one file per partition value.
After filtering heavily → redistribute remaining data.

`coalesce()`

df.coalesce(1)      # Merge into 1 partition (careful — single task!)
df.coalesce(10)     # Merge 200 → 10 without full shuffle

When to use:

Reducing small files before writing to HDFS/S3:

df.coalesce(10).write.parquet("s3://bucket/output/")

After heavy filtering where most partitions are empty.
Avoiding full shuffle when you just need fewer partitions.

Coalesce Pitfall

coalesce(1) forces all data through one task → OOM risk + no parallelism. Use repartition(1) only when data is tiny, or better: write with coalesce(N) where N = target file count.

Decision Tree

Need MORE partitions?           → repartition()
Need EVEN distribution?         → repartition(col)
Need to fix SKEW?               → repartition(col)
Joining two DFs on a key?       → repartition(n, key) on both
Reducing partitions, no shuffle?→ coalesce()
Writing small files?            → coalesce(N)
Reducing to 1 file?             → coalesce(1) (if small) or repartition(1)

AQE Note (Spark 3.x)

With Adaptive Query Execution, Spark can auto-coalesce shuffle partitions:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

This often eliminates the need for manual coalesce() after shuffles.

Integrating PySpark with Snowflake – Best Practices

Connection Options

Option A — Snowflake Connector for Spark (recommended for bulk):

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("SnowflakeIntegration") \
    .config("spark.jars.packages",
            "net.snowflake:snowflake-jdbc:3.14.0,"
            "net.snowflake:spark-snowflake_2.12:2.12.0-spark_3.3") \
    .config("spark.snowflake.sfURL", "https://<account>.snowflakecomputing.com") \
    .config("spark.snowflake.sfUser", "<user>") \
    .config("spark.snowflake.sfPassword", "<password>") \
    .config("spark.snowflake.sfDatabase", "MY_DB") \
    .config("spark.snowflake.sfSchema", "MY_SCHEMA") \
    .config("spark.snowflake.sfWarehouse", "MY_WH") \
    .config("spark.snowflake.sfRole", "MY_ROLE") \
    .getOrCreate()

Option B — Snowpark for Python (modern, recommended for Databricks):

from snowflake.snowpark import Session
session = Session.builder.configs({
    "account": "...", "user": "...", "password": "...",
    "database": "MY_DB", "schema": "MY_SCHEMA", "warehouse": "MY_WH"
}).create()
df = session.table("MY_TABLE")

Reading from Snowflake

sf_options = {
    "sfURL": "...",
    "sfUser": "...",
    "sfPassword": "...",
    "sfDatabase": "MY_DB",
    "sfSchema": "MY_SCHEMA",
    "sfWarehouse": "MY_WH",
    "sfRole": "MY_ROLE"
}

df = spark.read \
    .format("snowflake") \
    .options(**sf_options) \
    .option("dbtable", "MY_TABLE") \
    .option("query", "SELECT * FROM MY_TABLE WHERE dt = '2024-01-01'") \
    .load()

Writing to Snowflake

df.write \
    .format("snowflake") \
    .options(**sf_options) \
    .option("dbtable", "TARGET_TABLE") \
    .option("mode", "overwrite") \
    .option("sfCopyOptions", "ON_ERROR=CONTINUE") \
    .save()

Best Practices

#	Practice	Why
1	Use `query` pushdown instead of `dbtable` when filtering	Snowflake computes filters; less data transferred
2	Use Snowflake’s warehouse for heavy aggregations, not Spark	Snowflake’s optimizer is often faster
3	Stage data in S3/GCS/Azure for large writes, then `COPY INTO`	The connector stages via internal stage — external stage is faster & cheaper
4	Use key-pair or OAuth auth, not passwords, in production	Security & no rotation headaches
5	Set `sfWarehouse` to a dedicated ETL warehouse	Avoid contention with BI queries
6	Tune `sfParallelism` and `sfBatchSize`	Match Snowflake warehouse size
7	Use `sfRole` with least privilege	Security
8	Partition writes with `sfCreateTableOptions`	Better pruning
9	Use Snowpark if on Databricks — native, no connector overhead	Performance
10	Prefer `COPY INTO` over row-by-row inserts	10–100x faster
11	Monitor `QUERY_HISTORY` in Snowflake	Find slow queries
12	Use `spark.snowflake.sfCompression` = gzip	Less network I/O
13	Avoid `collect()` on Snowflake reads	Driver OOM
14	Use Snowflake’s `VARIANT` for semi-structured data, not Spark’s `explode` in a loop	Pushdown
15	Cache nothing across sessions — use Snowflake’s result cache	Cheaper

Performance Tips

# 1. Filter in Snowflake, not Spark
df = spark.read.format("snowflake").options(**sf_options) \
    .option("query", "SELECT * FROM T WHERE dt >= '2024-01-01'") \
    .load()

# 2. Push aggregations down
.option("query", "SELECT country, SUM(amt) FROM T GROUP BY country")

# 3. For large writes, use external stage + COPY INTO
df.write.format("snowflake").options(**sf_options) \
    .option("dbtable", "TARGET") \
    .option("sfCopyOptions", "ON_ERROR=CONTINUE PURGE=TRUE") \
    .mode("overwrite").save()

PySpark Code: Join Two DataFrames + Window Functions for Ranking

Problem Statement

employees: emp_id, name, dept_id, salary
departments: dept_id, dept_name, location
Task: Join them, then rank employees by salary within each department.

Complete Code

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, rank, dense_rank, row_number, percent_rank,
    ntile, sum, avg, max, when, desc
)
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("JoinAndRank").getOrCreate()

# ---------- Sample Data ----------
employees_data = [
    (1, "Alice",   10, 90000),
    (2, "Bob",     10, 85000),
    (3, "Charlie", 10, 85000),
    (4, "Diana",   20, 120000),
    (5, "Eve",     20, 95000),
    (6, "Frank",   20, 70000),
    (7, "Grace",   30, 110000),
    (8, "Heidi",   30, 110000),
    (9, "Ivan",    30, 60000),
]
employees = spark.createDataFrame(
    employees_data, ["emp_id", "name", "dept_id", "salary"]
)

departments_data = [
    (10, "Engineering", "NYC"),
    (20, "Data Science", "SF"),
    (30, "Product", "Austin"),
]
departments = spark.createDataFrame(
    departments_data, ["dept_id", "dept_name", "location"]
)

# ---------- 1. JOIN ----------
joined = employees.join(departments, on="dept_id", how="inner")

# ---------- 2. WINDOW SPEC ----------
# Partition by department, order by salary descending
window_spec = Window.partitionBy("dept_id").orderBy(col("salary").desc())

# ---------- 3. APPLY RANKING FUNCTIONS ----------
result = joined.select(
    col("emp_id"),
    col("name"),
    col("dept_name"),
    col("location"),
    col("salary"),
    rank().over(window_spec).alias("rank"),                    # ties → same rank, gaps
    dense_rank().over(window_spec).alias("dense_rank"),        # ties → same rank, no gaps
    row_number().over(window_spec).alias("row_num"),           # unique, no ties
    percent_rank().over(window_spec).alias("pct_rank"),        # 0 to 1
    ntile(3).over(window_spec).alias("salary_quartile"),       # bucket into 3
    sum("salary").over(Window.partitionBy("dept_id")).alias("dept_total"),
    avg("salary").over(Window.partitionBy("dept_id")).alias("dept_avg"),
    max("salary").over(Window.partitionBy("dept_id")).alias("dept_max"),
)

result.orderBy("dept_id", "rank").show(truncate=False)

# ---------- 4. FILTER: Top 2 earners per department ----------
top2 = result.filter(col("rank") <= 2) \
             .select("dept_name", "name", "salary", "rank") \
             .orderBy("dept_name", "rank")
top2.show()

Expected Output

+------+-------+------------+--------+------+----+----------+--------+--------+----------+-------------+----------+--------+
|emp_id|name   |dept_name   |location|salary|rank|dense_rank|row_num |pct_rank|salary_...|dept_total|dept_avg|dept_max|
+------+-------+------------+--------+------+----+----------+--------+--------+----------+----------+--------+--------+
|1     |Alice  |Engineering |NYC     |90000 |1   |1         |1       |0.0     |1         |260000    |86666.67|90000   |
|2     |Bob    |Engineering |NYC     |85000 |2   |2         |2       |0.5     |2         |260000    |86666.67|90000   |
|3     |Charlie|Engineering |NYC     |85000 |2   |2         |3       |0.5     |3         |260000    |86666.67|90000   |
|4     |Diana  |Data Science|SF      |120000|1   |1         |1       |0.0     |1         |285000    |95000.00|120000  |
|5     |Eve    |Data Science|SF      |95000 |2   |2         |2       |0.5     |2         |285000    |95000.00|120000  |
|6     |Frank  |Data Science|SF      |70000 |3   |3         |3       |1.0     |3         |285000    |95000.00|120000  |
|7     |Grace  |Product     |Austin  |110000|1   |1         |1       |0.0     |1         |280000    |93333.33|110000  |
|8     |Heidi  |Product     |Austin  |110000|1   |1         |2       |0.0     |2         |280000    |93333.33|110000  |
|9     |Ivan   |Product     |Austin  |60000 |3   |2         |3       |1.0     |3         |280000    |93333.33|60000   |
+------+-------+------------+--------+------+----+----------+--------+--------+----------+----------+--------+--------+

Key Concepts

Function	Behavior on Ties	Use Case
`rank()`	Same rank, gaps (1,2,2,4)	Competition ranking
`dense_rank()`	Same rank, no gaps (1,2,2,3)	Leaderboards
`row_number()`	Always unique (1,2,3,4)	Deduplication
`percent_rank()`	(rank-1)/(n-1)	Percentile position
`ntile(n)`	Buckets into n groups	Deciles, quartiles
`lag()/lead()`	Access prev/next row	YoY comparisons
`sum().over()`	Running total	Cumulative metrics

Common Patterns

# Running total
Window.partitionBy("dept_id").orderBy("salary").rowsBetween(Window.unboundedPreceding, Window.currentRow)

# Deduplication — keep latest record per key
Window.partitionBy("emp_id").orderBy(col("updated_at").desc())
df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")

# Moving average (3-row window)
Window.partitionBy("dept_id").orderBy("salary").rowsBetween(-1, 1)

Transformation Tasks in Notepad (No IDE) — Logic & Syntax

This tests whether you can write correct PySpark on paper without autocomplete, syntax highlighting, or a REPL. Here are the common tasks and how to nail them.

Tips for Notepad Coding

Memorize imports — from pyspark.sql.functions import col, when, lit, ...
Write column references as col("name") — never df.name (fragile).
Use alias() after every transformation to keep schemas readable.
Indent consistently — 4 spaces.
Chain with \ or wrap in parentheses for multi-line.
Comment each step — interviewers value clarity.
Handle nulls explicitly — when(col("x").isNull(), lit(0)).otherwise(col("x")).

Task 1: Deduplicate — Keep Latest Record per Key

from pyspark.sql.functions import col, row_number
from pyspark.sql.window import Window

w = Window.partitionBy("customer_id").orderBy(col("updated_at").desc())

deduped = df.withColumn("rn", row_number().over(w)) \
            .filter(col("rn") == 1) \
            .drop("rn")

Task 2: Pivot — Rows to Columns

pivoted = df.groupBy("customer_id") \
            .pivot("month") \
            .agg(sum("amount"))

Task 3: Handle Nulls — Replace with Default

from pyspark.sql.functions import coalesce, lit

cleaned = df.withColumn("amount", coalesce(col("amount"), lit(0))) \
            .withColumn("name", when(col("name").isNull(), lit("Unknown"))
                                .otherwise(col("name")))

Task 4: Parse JSON Column

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True)
])

parsed = df.withColumn("parsed", from_json(col("json_col"), schema)) \
           .select("parsed.*")

Task 5: Word Count (RDD style)

counts = sc.textFile("input.txt") \
    .flatMap(lambda line: line.split(" ")) \
    .map(lambda word: (word, 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .sortBy(lambda x: x[1], ascending=False)

Task 6: Word Count (DataFrame style)

from pyspark.sql.functions import explode, split, col

counts = df.select(explode(split(col("line"), " ")).alias("word")) \
           .groupBy("word").count() \
           .orderBy(col("count").desc())

Task 7: Join + Aggregate + Filter

from pyspark.sql.functions import sum, col

result = orders.join(customers, "customer_id") \
    .groupBy("customer_id", "customer_name") \
    .agg(sum("amount").alias("total_spent")) \
    .filter(col("total_spent") > 1000) \
    .orderBy(col("total_spent").desc())

Task 8: Add Sequential ID

from pyspark.sql.functions import monotonically_increasing_id

df_with_id = df.withColumn("id", monotonically_increasing_id())

Task 9: Conditional Column

from pyspark.sql.functions import when, col

df = df.withColumn(
    "tier",
    when(col("amount") > 1000, "Gold")
    .when(col("amount") > 500, "Silver")
    .otherwise("Bronze")
)

Task 10: Explode Array Column

from pyspark.sql.functions import explode, explode_outer

df.select("id", explode("items").alias("item"))
# explode_outer keeps null/empty arrays

Common Syntax Pitfalls to Avoid

Mistake	Correct
`df.select(df.name)`	`df.select(col("name"))`
`df.filter("age > 30")`	`df.filter(col("age") > 30)` (safer)
`df.groupBy("x").sum("y")`	`df.groupBy("x").agg(sum("y").alias("total"))`
`df.withColumn("new", "old")`	`df.withColumn("new", col("old"))`
Forgetting `Window` import	`from pyspark.sql.window import Window`
`df.dropDuplicates()` on all cols	`df.dropDuplicates(["id"])` for subset
`count()` on `groupBy`	Use `.count()` not `.agg(count("*"))` unnecessarily

Interview Strategy

Restate the problem — “You want me to find the top N per group, right?”
Sketch the plan — “I’ll use a window function partitioned by X, ordered by Y.”
Write imports first — saves time.
Write the code with clear variable names.
Test mentally with a small example.
Mention edge cases — nulls, duplicates, empty partitions.
Mention optimizations — broadcast, AQE, repartition.

Summary Cheat Sheet

Topic	One-Liner
Salting	Add random suffix to hot keys to spread skew across partitions
Bloom Filter	Probabilistic set membership → prunes shuffle I/O in joins
Spark Memory	60% Spark (50/50 exec/storage) + 40% user; G1GC + Kryo
SparkContext vs SparkSession	SC = RDD entry point; SS = unified entry (SQL + DF + RDD)
Repartition vs Coalesce	Repartition = shuffle, even, can increase; Coalesce = no shuffle, decrease only
Snowflake + PySpark	Use connector or Snowpark; push filters/aggregations down; key-pair auth
Window Functions	`rank`, `dense_rank`, `row_number`, `ntile`, `lag/lead` over `Window.partitionBy().orderBy()`
Notepad Coding	Memorize imports, use `col()`, `alias()`, handle nulls, comment steps

How does Delta Lake ensure ACID transactions on big data?

Delta Lake provides ACID guarantees on top of object storage (S3, ADLS, GCS) using a transaction log (_delta_log) and optimistic concurrency control.
- Atomicity — Every write (INSERT, UPDATE, DELETE, MERGE) is recorded as a single commit in the Delta log. If the job fails midway, the partial files are never referenced by the log, so readers never see a broken state.
- Consistency — Schema is enforced on write. If a column is INT, inserting a string fails rather than corrupting the table.
- Isolation — Multiple readers and writers can hit the same table concurrently. Delta uses optimistic concurrency: writers validate against the latest log snapshot at commit time and retry on conflict. Readers always see a consistent snapshot.
- Durability — Once a commit is in the log, it’s permanent. If a cluster dies, the table state is rebuilt by replaying the JSON + Parquet checkpoints in _delta_log.
How does Databricks implement streaming pipelines using Structured Streaming?

Databricks uses Apache Spark Structured Streaming, which treats a stream as an unbounded table and processes it as incremental micro-batches.
- Ingestion — Auto Loader (cloudFiles) incrementally discovers new files in cloud storage without listing the whole bucket. Delta tables can also be a streaming source/sink, giving exactly-once semantics.
- Processing — Transformations use the same DataFrame API as batch. Stateful ops (aggregations, stream-stream joins, dedup) require watermarks to bound late data.
- Fault tolerance — Checkpoints store offsets and state so a failed query resumes exactly where it stopped (exactly-once).
- Latency control — trigger(processingTime=...) for micro-batch, trigger(availableNow=True) for incremental batch, or Real-Time Mode for ~millisecond latency.
- Medallion pattern — Typical layout: Bronze (raw ingest) → Silver (cleaned/conformed) → Gold (aggregated for BI), each stage a Delta table.
What are Z-ordering and data skipping in Databricks, and how do they improve performance?
- Data skipping — Delta Lake stores per-file column statistics (min, max, null count) in the log. When a query has a filter like WHERE date = '2024-01-01', the engine skips any file whose min/max range can’t contain that value. Less I/O = faster queries.
- Z-ordering — Data skipping only helps if related data is clustered. OPTIMIZE table ZORDER BY (col) physically co-locates rows with similar values into the same files, so the min/max ranges become tight and skipping becomes effective.
- Caveat — Z-ordering is expensive on write and doesn’t help high-cardinality columns. Databricks now recommends Liquid Clustering for new tables because it’s incremental (no full rewrite) and lets you change clustering keys.
How would you optimize a slow-performing Databricks job (tuning clusters, caching, partitioning)?

Clusters
- Enable autoscaling and auto-termination.
- Right-size executor memory — too small causes GC pressure/OOM, too large wastes DBUs.
- Use Photon for SQL/DataFrame workloads.
Caching
- .cache() / .persist(MEMORY_AND_DISK) when the same DataFrame is reused across multiple actions.
- Don’t cache one-shot DataFrames — it just adds overhead.
Partitioning & layout
- Target ~128 MB per partition; use repartition() / coalesce().
- Fix skew with AQE (spark.sql.adaptive.skewJoin.enabled=true) or salting.
- broadcast() small tables in joins to avoid shuffles.
- Run OPTIMIZE to compact small files.
Code
- Enable Adaptive Query Execution (AQE).
- Avoid collect() to the driver and row-by-row Python UDFs — use built-in functions.
Explain how Databricks Unity Catalog helps with governance and security in multi-tenant environments.

Unity Catalog is a centralized metastore + permission model spanning all workspaces in an account.
- 3-level namespace — Catalog.Schema.Table/View/Volume, so you grant at any level.
- Storage credentials — Access to cloud storage is via a managed credential (Azure Access Connector, AWS IAM role, GCP service account). Users get data permissions, never raw storage keys.
- Fine-grained access — Table-, column-, and row-level security; dynamic views; attribute-based policies.
- Audit & lineage — Every access is logged; column-level lineage is tracked automatically.
- Tenant isolation — Each tenant/team gets its own Catalog or Schema with scoped grants, so cross-tenant leakage is prevented by design.
- Delta Sharing — Share data across orgs without copying it.
How would you design a cost-optimized Databricks architecture for a production workload?

Compute
- Job clusters for ETL (cheaper than all-purpose).
- Serverless SQL Warehouses for BI — scale to zero, pay per query.
- Photon on SQL/DataFrame jobs — faster jobs usually cost less overall.
- Spot/preemptible instances for fault-tolerant batch; on-demand for critical SLAs.
- Instance pools to cut cluster start time.
Elasticity
- Autoscaling + aggressive auto-termination (e.g., 30 min idle).
- Compute policies to cap max workers and forbid oversized instance types.
Data
- Medallion architecture with Delta Lake.
- Liquid Clustering instead of static partitions where cardinality is high.
- Scheduled OPTIMIZE + VACUUM (7-day default retention) to control storage and file count.
- Tier cold data to cheaper storage classes.
Governance
- Tag workspaces, clusters, and warehouses by team/project/env for chargeback.
- Use system tables (system.billing.usage) to monitor DBU spend and catch runaway jobs.\
What is PySpark, and why is it used in Data Engineering?

PySpark is the Python API for Apache Spark, a distributed computing framework designed for processing large-scale data across a cluster of machines. It lets you write Python code that Spark translates into distributed jobs running in parallel across many nodes.

It’s used in Data Engineering because:
- Scale — Handles terabytes to petabytes of data that single-machine tools like pandas can’t touch.
- Unified engine — Batch, streaming, SQL, ML, and graph processing all in one framework.
- Python ecosystem — Integrates with pandas, NumPy, scikit-learn, Airflow, and the wider Python data stack.
- Fault tolerance — Lineage-based recovery means failed tasks are recomputed automatically.
- Lakehouse fit — Native integration with Delta Lake, Iceberg, and Hudi for ACID tables.
What is the difference between RDD, DataFrame, and Dataset in Spark?

RDD (Resilient Distributed Dataset) is the lowest-level abstraction — an immutable, distributed collection of objects with no schema. You get full control but no Catalyst optimization, and it’s verbose to write. Mostly used for unstructured data or when you need fine-grained control.

DataFrame is a distributed collection organized into named columns with a schema. It sits on top of RDDs but goes through the Catalyst optimizer and Tungsten execution engine, so it’s much faster. It’s the default choice for structured data.

Dataset is a typed, object-oriented API — available in Scala and Java, not Python. It combines DataFrame’s optimization with compile-time type safety. In PySpark, you effectively only have RDDs and DataFrames.
What is SparkSession, and why is it required?

SparkSession is the entry point to Spark in modern versions (2.0+). It replaced the older SparkContext, SQLContext, and HiveContext with a single unified object.

It’s required because:
- It’s how you configure and connect to the cluster (appName, master, configs).
- It gives you access to spark.read, spark.sql, spark.createDataFrame, and catalog operations.
- It manages the underlying SparkContext and session state.
- In Databricks, a SparkSession is auto-created as spark, but you can still create a new one with .builder.
What is lazy evaluation in PySpark?

Lazy evaluation means Spark doesn’t execute transformations the moment you write them. It builds a logical plan (a DAG of operations) and only runs the computation when you trigger an action like show(), count(), collect(), or write().

Why it matters:
- Optimization — Catalyst can reorder, combine, and prune the whole plan before execution.
- Efficiency — If you filter early, Spark may push that filter down to the data source and never read unnecessary data.
- Fewer passes — Multiple transformations can be fused into a single stage.
What is the difference between transformations and actions?

Transformations are lazy operations that return a new RDD/DataFrame — select(), filter(), withColumn(), join(), groupBy(). They don’t run until an action is called.

Actions trigger execution and return a result to the driver or write to storage — show(), count(), collect(), take(), write(), foreach().

Rule of thumb: if it returns a DataFrame, it’s a transformation. If it returns a value, a list, or writes data, it’s an action.
What is the difference between narrow and wide transformations?

Narrow transformations — Each output partition depends on exactly one input partition. No data moves across the network. Examples: map(), filter(), select(), withColumn(), union().

Wide transformations — Output partitions depend on multiple input partitions, so data must be shuffled across the network. Examples: groupBy(), join() (non-broadcast), distinct(), repartition(), orderBy().

Narrow transformations are cheap and stay within a stage. Wide transformations are expensive, create stage boundaries, and are usually where jobs slow down.
What is a shuffle in Spark, and why can it impact performance?

A shuffle is the process of redistributing data across partitions so that rows with the same key end up on the same executor. It happens during wide transformations.

It hurts performance because:
- Disk I/O — Shuffle data is written to disk on the map side and read back on the reduce side.
- Network I/O — Data is transferred between executors, which is far slower than local memory access.
- Serialization — Everything gets serialized and deserialized.
- Skew sensitivity — If one key dominates, one task does all the work while others idle.
Minimizing shuffles is one of the biggest levers for PySpark performance.
What is the difference between repartition() and coalesce()?

repartition(n) — Full shuffle. Can increase or decrease partitions and can partition by column. Expensive but produces evenly sized partitions.

coalesce(n) — Narrow transformation. Only decreases partitions by merging existing ones. No shuffle, so it’s cheap — but it can create uneven partitions.

Use repartition() when you need to increase parallelism, fix skew, or partition by a key before writing. Use coalesce() when you just want fewer output files after a filter reduced the data size.
What is data skew, and how do you handle it in PySpark?

Data skew is when a small number of partitions hold a disproportionately large share of the data — usually because one key value (like null or a dominant customer ID) appears millions of times.

Symptoms: most tasks finish in seconds, a few run for hours; Spark UI shows one task with huge shuffle read.

How to handle it:
- Salting — Add a random suffix to the hot key, aggregate, then re-aggregate without the suffix.
- Broadcast join — If the other side is small, broadcast it to avoid the shuffle entirely.
- AQE skew join — Enable spark.sql.adaptive.skewJoin.enabled=true; Spark splits oversized partitions automatically.
- Filter or isolate hot keys — Process them separately and union the result.
- Repartition by a better key — If the join key is skewed, join on a composite key.
What is a broadcast join, and when would you use it?

A broadcast join sends a small table to every executor so the join happens locally, with no shuffle of the large table.

You use it when:
- One side is small enough to fit in memory (default threshold ~10 MB, tunable via spark.sql.autoBroadcastJoinThreshold).
- You’re joining a large fact table with a small dimension table.
In PySpark:
python
```
from pyspark.sql.functions import broadcast
large_df.join(broadcast(small_df), "id")
```
Don’t broadcast tables that are too large — you’ll OOM the executors.
How do you optimize a slow PySpark job?
- Reduce shuffles — Filter and select early, broadcast small tables, avoid unnecessary repartition().
- Fix skew — Salting, AQE skew join, or isolating hot keys.
- Tune partitions — Target ~128 MB per partition; too few means idle cores, too many means task overhead.
- Cache wisely — Persist DataFrames reused across multiple actions.
- Enable AQE — spark.sql.adaptive.enabled=true for runtime plan adjustments.
- Use built-in functions — Avoid Python UDFs where possible.
- Column pruning — Select only needed columns, especially with Parquet/Delta.
- Predicate pushdown — Filter on partition columns so files are skipped.
- File layout — Compact small files with OPTIMIZE; use Delta or Parquet.
- Resource tuning — Right-size executor memory/cores, enable dynamic allocation.
- Check the Spark UI — Look at stage timelines, shuffle read/write, and task duration distribution.
What is the difference between cache() and persist()?

cache() is shorthand for persist(StorageLevel.MEMORY_AND_DISK) in PySpark — it stores the DataFrame in memory, spilling to disk if needed.

persist(level) lets you choose the storage level explicitly:
- MEMORY_ONLY — Fastest, but recomputes if it doesn’t fit.
- MEMORY_AND_DISK — Default for cache(), spills to disk.
- DISK_ONLY — Slower but memory-friendly.
- MEMORY_ONLY_SER / MEMORY_AND_DISK_SER — Serialized, less memory but more CPU.
Both are lazy — data is materialized on the first action. Always unpersist() when done to free memory.
How do you handle duplicate records in PySpark?
- Drop all duplicates — df.dropDuplicates() or df.distinct().
- Drop duplicates on specific columns — df.dropDuplicates(["id", "date"]) keeps the first occurrence.
- Keep the latest record — Use a window function:
python
```
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, col

w = Window.partitionBy("id").orderBy(col("updated_at").desc())
df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")
```
- Aggregate — groupBy("id").agg(max("updated_at")) when you only need one value per key.
- Delta MERGE — For upserts into a Delta table, MERGE handles dedup on write.
How do you handle NULL values in PySpark?
- Detect — df.filter(col("x").isNull()), or df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).
- Drop — df.dropna(), df.dropna(subset=["col1"]), df.dropna(thresh=2).
- Fill — df.fillna(0), df.fillna({"col1": 0, "col2": "unknown"}).
- Replace — df.replace(None, "N/A") or df.na.replace().
- Coalesce — coalesce(col("a"), col("b"), lit(0)) to take the first non-null.
- Null-safe joins — Use eqNullSafe (<=>) when NULL should match NULL.

How do you use Window Functions in PySpark?

Window functions compute a value per row based on a group of related rows, without collapsing them like groupBy().

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, lag, sum

w = Window.partitionBy("dept").orderBy("salary")

df.withColumn("rank", rank().over(w)) \
  .withColumn("prev_salary", lag("salary", 1).over(w)) \
  .withColumn("running_total", sum("salary").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow)))

Key parts:

partitionBy — Groups rows (like GROUP BY).
orderBy — Orders within each partition.
rowsBetween / rangeBetween — Defines the frame.

Common uses: ranking, running totals, previous/next row comparisons, dedup.

What is the difference between groupBy() and Window Functions?

groupBy() collapses rows — you get one row per group, and non-aggregated columns are lost. Use it for summary statistics.

Window functions keep every row and add a computed column. Use them for rankings, running totals, lag/lead comparisons, and dedup — anything where you need row-level context.

Example: groupBy("dept").avg("salary") gives one row per department. avg("salary").over(Window.partitionBy("dept")) gives every employee their department’s average alongside their own row.
What is a UDF, and why can UDFs affect PySpark performance?

A UDF (User-Defined Function) lets you run custom Python logic on a column:
python
```
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def clean(s):
    return s.strip().lower() if s else None

clean_udf = udf(clean, StringType())
df.withColumn("clean", clean_udf("raw"))
```
They hurt performance because:
- JVM ↔ Python serialization — Every row is pickled to Python and back, which is very slow.
- Block Catalyst — The optimizer can’t see inside a UDF, so no predicate pushdown or codegen.
- Row-by-row — No vectorization.
Better alternatives:
- Built-in functions — trim(), lower(), regexp_replace().
- Pandas UDFs (vectorized) — @pandas_udf processes data in Arrow batches, 10–100x faster.
- SQL expressions — Often express the same logic without leaving the JVM.
How do you read and write large datasets efficiently in PySpark?

Reading
- Use Parquet or Delta, not CSV/JSON — columnar, compressed, splittable.
- Column pruning — Select only needed columns.
- Predicate pushdown — Filter on partition columns so files are skipped.
- Partitioned reads — Store data partitioned by date/region and read only relevant partitions.
- Auto Loader — For incremental file ingestion.
Writing
- Control file count — Target ~128 MB per file; use repartition() or coalesce() before write.
- Partition by low-cardinality columns — Avoid over-partitioning (thousands of tiny dirs).
- Use Delta — Gives ACID, schema evolution, and OPTIMIZE for compaction.
- Overwrite vs append — Use mode("overwrite") with replaceWhere for targeted rewrites.
- Compression — snappy (default), zstd for better ratio.
How would you process a 1 TB dataset using PySpark?
- Store it as Parquet/Delta, partitioned by a sensible column (e.g., date).
- Read only what you need — filter and select early so Catalyst prunes files and columns.
- Right-size the cluster — Enough executors to keep all cores busy, with memory for shuffle.
- Target ~128 MB partitions — Repartition if needed.
- Broadcast small dimension tables, and let AQE handle skew on big joins.
- Cache only reused DataFrames — Don’t cache the whole 1 TB.
- Write partitioned output — Avoid one giant file or thousands of tiny ones.
- Monitor in Spark UI — Check for skew, spill, and shuffle read/write sizes.
- Use Delta MERGE for incremental updates instead of full rewrites.
- Consider Databricks Photon — Faster SQL/DataFrame execution on the same data.
How do you troubleshoot a failed or slow PySpark job?

Start with the Spark UI
- Stages tab — Find the slowest stage; check shuffle read/write and spill.
- Tasks tab — Look at task duration distribution. One long task = skew.
- SQL tab — Inspect the physical plan; look for Cartesian products or missing pushdown.
- Executors tab — Check for OOM, GC time, and uneven task distribution.
Common causes and fixes
- OOM on driver — You called collect() on a huge dataset; write to storage instead.
- OOM on executor — Increase executor memory, reduce partition size, or fix skew.
- Skew — Salt the key, broadcast, or enable AQE skew join.
- Too many small tasks — coalesce() or repartition() to fewer, larger partitions.
- Too few partitions — Underutilized cluster; repartition up.
- Shuffle spill — Increase spark.sql.shuffle.partitions or executor memory.
- Slow UDFs — Replace with built-in functions or Pandas UDFs.
- Serialization errors — Check for non-serializable objects captured in closures.
- Data source bottleneck — Too many small files; compact with OPTIMIZE.
Logs
- Driver logs show the exception stack trace.
- Executor logs show the actual failure on the worker side.
- In Databricks, check the driver logs, Spark UI, and cluster event log.
How would you optimize a job processing 500 million records?

Optimizing at this scale is about reducing I/O, shuffles, and skew — not just adding more machines. Throwing hardware at a badly designed job usually just makes it fail faster.

First: understand the workload

Is it a one-time batch, a daily incremental, or a streaming job?

What’s the record size? 500M rows at 1 KB each = 500 GB; at 100 bytes = 50 GB. Very different problems.

Is it a full scan, a join-heavy job, or an aggregation?

What’s the SLA — minutes or hours?

Storage and file layout

Store data as Parquet or Delta, never CSV/JSON. Columnar + compressed + splittable.

Compact small files with OPTIMIZE — thousands of tiny files kill performance because each one is a separate task.

Partition by a low-cardinality, frequently-filtered column like date or region. Avoid over-partitioning (don’t partition by user_id).

Z-order or liquid clustering on commonly filtered/joined columns so data skipping works.

Target ~128 MB per file on write.

Read path

Column pruning — select only the columns you need.

Predicate pushdown — filter on partition columns so entire directories are skipped.

Read only the partitions you need — WHERE date BETWEEN ... instead of full table scans.

If ingesting incrementally, use Auto Loader or Delta MERGE instead of reprocessing everything.

Reduce shuffles

Broadcast joins for small dimension tables (under ~10 MB, tunable).

Bucket large tables on the join key so future joins avoid shuffles entirely.

Pre-aggregate before joins where possible — reduce row count before the expensive step.

Avoid distinct(), groupByKey(), and unnecessary repartition() — all trigger full shuffles.

Handle skew

Enable AQE (spark.sql.adaptive.enabled=true) and skew join (spark.sql.adaptive.skewJoin.enabled=true).

Salt hot keys if AQE isn’t enough.

Isolate the skewed keys, process them separately, and union back.

Watch for null keys — they all land on one partition.

Partitioning and parallelism

Target ~128 MB per partition; too few means idle cores, too many means task overhead.

Set spark.sql.shuffle.partitions based on data size, not the default 200. A rough guide: total_shuffle_bytes / 128 MB.

Use dynamic allocation so executors scale with the actual workload.

Caching

Only cache DataFrames that are reused across multiple actions. Caching a one-shot DataFrame is pure overhead.

Prefer persist(MEMORY_AND_DISK) for large data; MEMORY_ONLY for smaller reused sets.

Always unpersist() when done.

Code-level

Use built-in functions instead of Python UDFs — UDFs serialize row-by-row across JVM↔Python.

If you must use custom logic, use Pandas UDFs (Arrow-vectorized).

Avoid collect() on large data — it pulls everything to the driver and OOMs.

Avoid count() just for logging — it forces a full pass.

Filter and select as early as possible in the pipeline.

Cluster sizing

Match instance type to workload — memory-optimized for joins/aggregations, compute-optimized for simple transforms.

Right-size executor memory: too small causes GC/OOM, too large wastes resources and causes long GC pauses.

Use Photon on Databricks for SQL/DataFrame workloads — often 2–3x faster.

Use spot instances for fault-tolerant batch stages to cut cost.

Write path

Control output file count — repartition() or coalesce() before write.

Partition output by the column most queried downstream.

Use replaceWhere for targeted overwrites instead of full rewrites.

Compress with snappy (fast) or zstd (smaller).

Monitoring

Spark UI — check the slowest stage, shuffle read/write, spill, and task duration spread.

One task taking 10x longer than others = skew.

High shuffle spill = not enough memory or too-large partitions.

Long GC time = executor memory misconfigured.

Realistic outcome
A well-tuned job on 500M records usually runs in minutes to a couple of hours on a modest cluster. A poorly tuned one can take all day, spill constantly, and still fail with OOM. The difference is almost always in file layout, shuffle reduction, and skew handling — not raw cluster size.

Without Salting	With Salting
1 partition gets 10M rows	10 partitions get ~1M rows each
1 task = 3 hours	All tasks = ~18 min each
Cluster underutilized	Full parallelism

Pro	Con
Tiny memory footprint	False positives (rare)
O(k) insert/query	Can’t delete elements
Massive I/O savings	Tunable accuracy vs. size

PySpark & Databricks Interview Questions and Answers

PySpark & Databricks Interview Questions and Answers

You have a job where 199 tasks finish in 2 minutes but 1 task takes 40 minutes. How would you diagnose and fix it?

Design a schema drift detection framework. What drift types does it catch, and what’s the remediation path?

Streaming job crashes mid-micro-batch after writing output but before checkpoint commits. What happens on restart?

You ran VACUUM with 0-hour retention and a long-running streaming read fails with “file not found.” Why?

Two source records for the same employee arrive in the same batch with different attributes. How does SCD Type 2 MERGE handle this?

A production job ran fine for months and suddenly starts spilling to disk. What changed?

Pipeline has no monitoring and silently drops 2% of records for weeks. How do you find it and prevent recurrence?

How do you read and interpret a query execution plan? Optimize a query joining two 500M-row tables taking 40 minutes.

How would you design incremental loading for a source with no reliable updated_at column?

What causes a Spark job to throw an OutOfMemoryError (OOM), and how do you troubleshoot it – driver vs. executor? How would you diagnose an executor OOM in the Spark UI – what do you look at first?

Explain the Delta transaction log mechanics – what happens on every commit? What are .crc files, and what happens if one gets deleted manually?

Auto Loader vs. COPY INTO – which scales better? Auto Loader typically needs separate paths per source; if that’s not allowed, what would you propose instead? Can a replay mechanism be built on top of Auto Loader? When would you use COPY INTO instead of Auto Loader?

What is Liquid Clustering, and how does it differ from Z-ORDER + partitioning? Can both be used on the same table?

What is Change Data Feed (CDF), and how would you use it to build an incremental Silver-to-Gold pipeline?

Explain watermarking in Structured Streaming and why it’s needed for stateful aggregations.

What happens when you execute a query in Databricks? What is Adaptive Query Execution (AQE), and what 3 things does it do at runtime?

Explain Deletion Vectors — what problem do they solve?

Explain how Databricks lineage gets created in Unity Catalog — how would you use it to assess the blast radius of a schema change?

What is Databricks Lakeflow / Delta Live Tables (DLT), and when would you choose it over hand-written PySpark jobs?

Explain broadcast join vs. sort-merge join vs. shuffle-hash join. When does Spark choose each?

Explain persist() vs. cache(), and the different StorageLevels. How does Spark’s Unified Memory Manager handle the conflict when execution memory needs space that’s currently occupied by cached (storage memory) data — what actually gets evicted, and when?

Deep Clone vs. Shallow Clone – note that renaming a table also renames the underlying storage directory.

Explain clustered vs. non-clustered indexing, and how a wrong index choice can make a query slower, not faster.

Explain a Databricks Service Principal – how do you create one, and what’s it used for vs. a normal user identity?

What is Salting and how does it help with Data Skew?

The Problem: Data Skew

What is Salting?

How It Works

Why It Helps

When to Use Salting

Alternative Fixes for Skew

Explain Bloom Filters and how they improve Join Performance

What is a Bloom Filter?

How It Works Internally

How It Improves Join Performance

Enabling It in Spark

Other Uses of Bloom Filters

Trade-offs

How does Spark Memory Management work — Heap, Caching, GC?

Spark Memory Model (Unified Memory Management, Spark 1.6+)

Key Configs

Unified Memory: Execution ↔ Storage

Caching Strategies

Garbage Collection (GC)

Common Memory Errors

Difference between SparkContext and SparkSession

Code Comparison

Key Takeaway

Explain Repartition vs Coalesce with Use Cases

Quick Comparison

repartition()

coalesce()

Coalesce Pitfall

Decision Tree

AQE Note (Spark 3.x)

Integrating PySpark with Snowflake – Best Practices

Connection Options

Reading from Snowflake

Writing to Snowflake

Best Practices

Performance Tips

PySpark Code: Join Two DataFrames + Window Functions for Ranking

Problem Statement

Complete Code

Expected Output

Key Concepts

Common Patterns

Transformation Tasks in Notepad (No IDE) — Logic & Syntax

Tips for Notepad Coding

Task 1: Deduplicate — Keep Latest Record per Key

Task 2: Pivot — Rows to Columns

Task 3: Handle Nulls — Replace with Default

Task 4: Parse JSON Column

Task 5: Word Count (RDD style)

Task 6: Word Count (DataFrame style)

Task 7: Join + Aggregate + Filter

Task 8: Add Sequential ID

Task 9: Conditional Column

`repartition()`

`coalesce()`