Claude Code for Spark Delta Lake (2026)
Delta Lake has become the backbone of modern data lakehouse architectures, providing ACID transactions, time travel, and schema enforcement on top of Apache Spark. But writing and maintaining Delta Lake pipelines can be complex. This tutorial shows you how Claude Code, a CLI-powered AI assistant, can dramatically improve your productivity when working with Spark and Delta Lake.
Why Use Claude Code for Data Engineering?
Claude Code isn’t just another code completion tool. It understands context, remembers your project structure, and can help you debug issues, generate boilerplate, and optimize your Delta Lake workflows. Whether you’re writing ETL pipelines, implementing data quality checks, or building streaming workflows, Claude Code acts as an intelligent pair programmer.
The key advantages include:
- Instant code generation for common Delta Lake patterns
- Debugging assistance with meaningful error interpretations
- Performance optimization suggestions for Spark jobs
- Documentation that keeps your team moving fast
When you describe your pipeline requirements in plain language. “merge today’s CDC records into the gold layer, handling late arrivals up to 48 hours”. Claude Code produces working PySpark code rather than forcing you to consult API docs for every merge condition variant.
Setting Up Your Development Environment
Before diving into Delta Lake workflows, ensure your environment is properly configured. Claude Code works best when it has access to your project’s context.
requirements.txt - Essential dependencies
pyspark>=3.4.0
delta-spark>=2.4.0
Initialize your Delta Lake project structure:
mkdir delta-lake-project
cd delta-lake-project
mkdir notebooks scripts tests
Giving Claude Code Your Project Context
Create a CONTEXT.md or paste the following block at the start of any Claude Code session. The more accurately you describe your environment, the better the generated code fits your actual infrastructure:
Environment:
- Spark 3.5.1 on Databricks Runtime 14.3 LTS
- Delta Lake 3.1.0
- Python 3.11
- Cloud storage: AWS S3 (s3a:// paths)
- Catalog: Unity Catalog, default catalog = main
- Table format: Delta (no Hive metastore fallback)
Conventions:
- All tables use three-part names: main.<schema>.<table>
- Partitioning key is always `event_date DATE`
- Primary merge key is `record_id STRING`
- All ETL functions return a dict with keys: rows_read, rows_written, rows_skipped
With this block in scope, Claude Code will produce Unity Catalog-aware three-part table names, correct s3a:// paths, and consistent return signatures without being prompted for each function.
Core Delta Lake Operations with Claude Code
Creating and Managing Tables
One of the most common tasks is creating Delta Lake tables. Here’s how Claude Code helps you write clean, production-ready code:
from pyspark.sql import SparkSession
from delta import DeltaTable
spark = SparkSession.builder \
.appName("DeltaLakeWorkflow") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
Create a Delta Lake table with schema enforcement
schema = "id INT, name STRING, created_at TIMESTAMP, value DOUBLE"
df = spark.createDataFrame([
(1, "alpha", "2026-01-15", 100.5),
(2, "beta", "2026-02-20", 250.75)
], schema)
df.write.format("delta") \
.mode("overwrite") \
.save("/delta-lake-project/data/source_table")
Claude Code can also help you add metadata, partitioning, and table properties:
Add table properties for governance
DeltaTable.createOrReplace(spark) \
.tableName("production_data") \
.addColumn("id", "INT", nullable=False) \
.addColumn("data", "STRING") \
.property("description", "Main production table") \
.property("pipelines", "etl-daily") \
.execute()
Table Creation: SQL vs Python API Comparison
Both the Python API and Spark SQL can create Delta tables. Use the comparison below to choose the right approach for your situation, then tell Claude Code which style your team prefers so it generates consistent code throughout the project.
| Approach | Best for | Limitation |
|---|---|---|
df.write.format("delta") |
Quick prototype, ad-hoc exploration | No column-level comments |
DeltaTable.createOrReplace() builder |
New tables with governance metadata | Verbose for wide schemas |
spark.sql("CREATE TABLE ...") |
Replicating DDL from another system | Harder to parameterize in Python |
Delta Live Tables (DLT) @dlt.table |
Declarative pipelines with lineage | Requires Databricks runtime |
Implementing Incremental Data Processing
Delta Lake’s ability to track changes makes incremental processing efficient. Here’s a practical pattern:
from delta.tables import DeltaTable
def incremental_upsert(source_path, target_table, merge_condition):
"""
Perform MERGE (upsert) operation between source and target Delta table.
"""
source_df = spark.read.format("delta").load(source_path)
target_df = DeltaTable.forName(spark, target_table)
target_df.alias("target").merge(
source_df.alias("source"),
merge_condition
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
print(f"Successfully merged data into {target_table}")
Usage with timestamp-based filtering
merge_condition = "target.id = source.id"
incremental_upsert("/staging/new_data", "production_data", merge_condition)
A More Complete MERGE with Soft-Delete Support
Production CDC pipelines typically need to handle inserts, updates, and deletes in one pass. Ask Claude Code to extend the basic merge into a full CDC handler:
from delta.tables import DeltaTable
from pyspark.sql import DataFrame
from pyspark.sql.functions import col, current_timestamp
import logging
logger = logging.getLogger(__name__)
def apply_cdc_merge(
spark,
source_df: DataFrame,
target_table: str,
key_cols: list,
op_col: str = "cdc_operation", # "I", "U", "D"
deleted_flag_col: str = "is_deleted"
) -> dict:
"""
Apply CDC records from source_df into a Delta target table.
Supports insert (I), update (U), and soft-delete (D) operations.
Returns row counts for observability.
"""
target = DeltaTable.forName(spark, target_table)
# Build join condition from composite key
join_expr = " AND ".join(
[f"target.{k} = source.{k}" for k in key_cols]
)
# Separate deletes from upserts
deletes_df = source_df.filter(col(op_col) == "D")
upserts_df = source_df.filter(col(op_col).isin("I", "U"))
rows_before = target.toDF().count()
# Apply soft deletes
if deletes_df.count() > 0:
target.alias("target").merge(
deletes_df.alias("source"),
join_expr
).whenMatchedUpdate(
set={deleted_flag_col: "true",
"updated_at": "current_timestamp()"}
).execute()
# Apply upserts
if upserts_df.count() > 0:
target.alias("target").merge(
upserts_df.alias("source"),
join_expr
).whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()
rows_after = target.toDF().count()
result = {
"rows_read": source_df.count(),
"rows_written": rows_after - rows_before,
"rows_skipped": 0,
}
logger.info("CDC merge complete: %s", result)
return result
Time Travel and Data Versioning
One of Delta Lake’s most powerful features is time travel. Claude Code can help you construct queries that use this capability:
Query previous versions of a Delta table
Using timestamp
df_v1 = spark.read.format("delta") \
.option("timestampAsOf", "2026-01-01") \
.load("/delta-lake-project/data/source_table")
Using version number
df_v2 = spark.read.format("delta") \
.option("versionAsOf", 5) \
.load("/delta-lake-project/data/source_table")
Compare versions to identify changes
from pyspark.sql.functions import col
current_df = spark.read.format("delta").load("/data/table")
previous_df = spark.read.format("delta").option("versionAsOf", 3).load("/data/table")
changes = current_df.join(previous_df, "id", "outer") \
.where(col("current.value") != col("previous.value"))
Audit Trail with DESCRIBE HISTORY
Before using time travel for rollback, inspect what actually changed. Claude Code can generate audit helpers like this one:
def get_table_history(spark, table_name: str, limit: int = 20) -> DataFrame:
"""
Return the operation history for a Delta table as a DataFrame.
Useful for incident investigation and compliance audits.
"""
return (
spark.sql(f"DESCRIBE HISTORY {table_name} LIMIT {limit}")
.select(
"version",
"timestamp",
"operation",
"operationParameters",
"userMetadata",
"engineInfo"
)
.orderBy("version", ascending=False)
)
def rollback_to_version(spark, table_name: str, target_version: int) -> None:
"""
Restore a Delta table to a previous version using RESTORE.
Always verify the target version in history before calling.
"""
history_df = get_table_history(spark, table_name)
versions = [row["version"] for row in history_df.collect()]
if target_version not in versions:
raise ValueError(
f"Version {target_version} not found in history for {table_name}"
)
logger.info("Restoring %s to version %d", table_name, target_version)
spark.sql(
f"RESTORE TABLE {table_name} TO VERSION AS OF {target_version}"
)
logger.info("Restore complete")
Pass a DESCRIBE HISTORY output snippet to Claude Code when diagnosing a data quality incident. It will read the operation timestamps, identify the bad write, and generate the RESTORE command with the correct version number.
Optimizing Spark Performance for Delta Lake
Claude Code excels at helping you optimize performance. Here are key strategies:
Partitioning Strategy
Create partitioned Delta table for query performance
df.write.format("delta") \
.partitionBy("year", "month", "day") \
.mode("overwrite") \
.save("/data/events")
Z-Order optimization for frequently filtered columns
DeltaTable.forPath(spark, "/data/events").optimize().executeZOrderBy("event_id", "customer_id")
Compaction and Data Skipping
Compact small files for better read performance
from delta.optimize import optimize
Run file compaction
spark.conf.set("spark.databricks.delta.optimize.enabled", "true")
Auto-compaction after writes
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")
Performance Tuning Decision Table
When you paste a slow query plan or a cluster usage screenshot into Claude Code and ask for help, it typically works through the following decision tree. Using it proactively before querying can save hours of tuning.
| Symptom | Likely cause | Recommended action |
|---|---|---|
| Full table scan on large table | Missing or wrong partition key | Re-partition by high-cardinality date column |
| Many small Parquet files | Streaming micro-batch or high-frequency appends | Enable auto-compaction or schedule OPTIMIZE hourly |
| Slow point lookups by entity ID | No Z-Order on lookup columns | Run OPTIMIZE ... ZORDER BY (entity_id) |
| Skewed partitions, some tasks 10x slower | Low-cardinality partition key | Switch to higher-cardinality key or use salting |
| Repeated full scans despite filters | Column stats stale or absent | Run ANALYZE TABLE ... COMPUTE STATISTICS |
| Join shuffle dominates stage | Two large tables joining without bucketing | Use broadcast join for the smaller side |
| Delta log reads slow at high version count | Transaction log accumulation | Run VACUUM and checkpoint the log |
Auto-Optimize and Liquid Clustering
Delta Lake 3.x introduced Liquid Clustering as a replacement for static partitioning. Ask Claude Code to help you migrate:
Enable Liquid Clustering on a new table (Delta Lake 3.x / DBR 13.3+)
spark.sql("""
CREATE TABLE main.analytics.events
CLUSTER BY (event_date, customer_id)
AS SELECT * FROM staging.events_raw
""")
On an existing table, add clustering incrementally
spark.sql("""
ALTER TABLE main.analytics.events
CLUSTER BY (event_date, customer_id)
""")
Run clustering. equivalent to OPTIMIZE but respects cluster keys
DeltaTable.forName(spark, "main.analytics.events").optimize().executeCompaction()
Liquid Clustering avoids the file explosion that static partitionBy causes on high-cardinality columns and removes the need for Z-Order. Tell Claude Code which Databricks Runtime version you are on and it will recommend Liquid Clustering for DBR 13.3+ or Z-Order for earlier releases.
Building Solid Data Pipelines
For production pipelines, incorporate error handling and monitoring:
from pyspark.sql.utils import AnalysisException
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_delta_write(df, path, mode="overwrite", partition_cols=None):
"""Safe Delta Lake write with validation and error handling."""
try:
# Data quality check
if df.count() == 0:
logger.warning(f"Empty DataFrame, skipping write to {path}")
return False
# Schema validation
df.printSchema()
# Write with transaction
writer = df.write.format("delta").mode(mode)
if partition_cols:
writer = writer.partitionBy(*partition_cols)
writer.save(path)
logger.info(f"Successfully wrote to {path}")
return True
except AnalysisException as e:
logger.error(f"Schema mismatch: {e}")
raise
except Exception as e:
logger.error(f"Write failed: {e}")
raise
Usage in pipeline
safe_delta_write(transformed_df, "/production/analytics",
mode="overwrite", partition_cols=["year", "month"])
Idempotent Pipeline Pattern
Production pipelines must be safe to re-run after a partial failure. Ask Claude Code to wrap any write operation in this idempotent helper:
from pyspark.sql import DataFrame
from pyspark.sql.functions import lit, current_timestamp
from delta.tables import DeltaTable
from typing import Optional
import hashlib
def idempotent_write(
spark,
df: DataFrame,
target_table: str,
run_id: str,
key_cols: list,
partition_cols: Optional[list] = None
) -> dict:
"""
Write df into target_table in an idempotent way.
Records tagged with run_id are deleted before re-insert,
so re-running the same run_id produces identical results.
"""
# Tag every row with this pipeline run
staged = df.withColumn("_run_id", lit(run_id)) \
.withColumn("_loaded_at", current_timestamp())
if DeltaTable.isDeltaTable(spark, f"SELECT 1 FROM {target_table}"):
# Delete rows from previous attempt with same run_id
spark.sql(f"""
DELETE FROM {target_table}
WHERE _run_id = '{run_id}'
""")
writer = staged.write.format("delta").mode("append")
if partition_cols:
writer = writer.partitionBy(*partition_cols)
writer.saveAsTable(target_table)
count = staged.count()
return {"rows_written": count, "run_id": run_id}
Tell Claude Code your orchestration tool (Airflow, Databricks Workflows, dbt) and the run ID source (DAG run ID, job run ID) and it will wire this pattern into your specific scheduler.
Testing Your Delta Lake Workflows
Writing tests ensures your pipelines work correctly:
import pytest
from delta import DeltaTable
def test_incremental_upsert():
"""Test that upsert correctly updates and inserts records."""
# Setup test data
test_path = "/tmp/test_delta"
# Initial data
initial_df = spark.createDataFrame([(1, "original")], ["id", "value"])
initial_df.write.format("delta").save(test_path)
# Merge new data
merge_df = spark.createDataFrame([(1, "updated"), (2, "new")], ["id", "value"])
delta_table = DeltaTable.forPath(spark, test_path)
delta_table.merge(merge_df, "target.id = source.id") \
.whenMatchedUpdate(set={"value": "source.value"}) \
.whenNotMatchedInsertAll() \
.execute()
# Verify results
result_df = spark.read.format("delta").load(test_path)
assert result_df.count() == 2
updated_row = result_df.filter(col("id") == 1).first()
assert updated_row["value"] == "updated"
Expanding the Test Suite with Data Quality Assertions
Functional tests confirm the merge logic works; data quality tests confirm the output meets business rules. Ask Claude Code to generate a quality suite alongside every ETL function:
import pytest
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, when, isnull
from delta import DeltaTable
import tempfile
import os
@pytest.fixture(scope="session")
def spark():
return (
SparkSession.builder
.master("local[2]")
.appName("delta-tests")
.config("spark.sql.extensions",
"io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
.getOrCreate()
)
@pytest.fixture
def tmp_delta_path(tmp_path):
return str(tmp_path / "test_table")
def test_no_duplicate_keys_after_merge(spark, tmp_delta_path):
"""Primary keys must be unique after any merge operation."""
initial = spark.createDataFrame([(1, "a"), (2, "b")], ["id", "val"])
initial.write.format("delta").save(tmp_delta_path)
updates = spark.createDataFrame([(1, "a_new"), (3, "c")], ["id", "val"])
DeltaTable.forPath(spark, tmp_delta_path) \
.alias("t").merge(updates.alias("s"), "t.id = s.id") \
.whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
result = spark.read.format("delta").load(tmp_delta_path)
dup_count = result.groupBy("id").count().filter(col("count") > 1).count()
assert dup_count == 0, f"Found {dup_count} duplicate keys"
def test_no_nulls_in_required_columns(spark, tmp_delta_path):
"""Non-nullable business columns must have zero nulls after write."""
data = spark.createDataFrame(
[(1, "alice", 99.9), (2, None, 50.0)],
["id", "name", "score"]
)
data.write.format("delta").save(tmp_delta_path)
result = spark.read.format("delta").load(tmp_delta_path)
null_names = result.filter(isnull(col("name"))).count()
assert null_names == 0, f"Found {null_names} null names in required column"
def test_row_count_within_expected_range(spark, tmp_delta_path):
"""Written row count must fall within an expected band (±20%)."""
rows = [(i, f"row_{i}", float(i)) for i in range(1000)]
df = spark.createDataFrame(rows, ["id", "name", "value"])
df.write.format("delta").save(tmp_delta_path)
actual = spark.read.format("delta").load(tmp_delta_path).count()
assert 800 <= actual <= 1200, f"Row count {actual} outside expected range"
Paste failing test output into Claude Code and ask it to fix the ETL function. It will trace the assertion failure back to the transformation logic rather than guessing.
Best Practices and Actionable Advice
- Always use schema enforcement: Let Delta Lake catch data quality issues early
- Implement proper partitioning: Balance file size (1GB target) with query patterns
- Enable column mapping: For schema evolution without table relocation
- Use Delta Live Tables: For declarative pipeline definitions when possible
- Use Unity Catalog: For enterprise governance across workspaces
Claude Code can help you refactor existing code to follow these patterns and suggest improvements specific to your use case.
Production Readiness Checklist
Before promoting any Delta Lake pipeline from development to production, walk through this checklist with Claude Code. Paste the checklist into your session and ask Claude Code to verify each item against your code:
| Checklist item | Why it matters |
|---|---|
| Schema enforcement enabled on target tables | Prevents silent column drops or type changes from upstream |
| All writes use transactions (Delta ACID, not raw Parquet) | Enables consistent reads during concurrent writes |
| OPTIMIZE scheduled after bulk loads | Prevents small-file proliferation that degrades read performance |
| VACUUM retention >= 7 days | Preserves time travel window for incident response |
| Row count and null-check assertions in pipeline | Catches bad data before it reaches downstream consumers |
| Idempotent write pattern or run_id tagging | Safe re-runs after partial failures |
| Delta change data feed enabled on CDC source tables | Required for incremental reads by downstream consumers |
| Table owner and comment set in Unity Catalog | Data discoverability and accountability |
| Alert on empty-DataFrame writes | Distinguishes “no new data” from “pipeline silently broken” |
Iterating with Claude Code on Performance Issues
The most effective way to use Claude Code for performance work is to paste the Spark UI stage summary or the EXPLAIN output directly into the session:
Here is the output of df.explain(True) for a slow merge job.
The target table has 500M rows partitioned by event_date.
The source has 2M rows for today only.
Why is the job scanning all 500M rows instead of pruning partitions?
Claude Code will read the physical plan, identify the missing predicate pushdown, and suggest the correct filter to add before the merge call. This workflow. paste real diagnostic output, ask a specific question. is faster than asking generically how to optimize merges.
By integrating Claude Code into your Spark Delta Lake development workflow, you gain a powerful assistant that understands data engineering patterns, helps debug issues quickly, and accelerates your path to production. Start small, automate repetitive tasks, and let AI handle the boilerplate while you focus on business logic.
Try it: Paste your error into our Error Diagnostic for an instant fix.
Related Reading
- Claude Code for Delta Lake Schema Evolution Workflow. detailed look into detecting, migrating, and auditing schema changes specifically
- Claude Code for Beginners: Complete Getting Started Guide
- Best Claude Skills for Developers in 2026
- Claude Skills Guides Hub
- Claude Code for Delta Lake Workflow Guide
Built by theluckystrike. More at zovo.one
Get started → Generate your project setup with our Project Starter.