Skip to main content ホーム クリエイター dicklesworthstone pi_agent_rust databricks-performance-tuning
databricks-performance-tuning Optimize Databricks cluster and query performance.
Use when jobs are running slowly, optimizing Spark configurations,
or improving Delta Lake query performance.
Trigger with phrases like "databricks performance", "spark tuning",
"databricks slow", "optimize databricks", "cluster performance".
インストールへ移動 Skills Marketplace コミュニティが作成したAIスキルを発見・探索
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/Dicklesworthstone/pi_agent_rust --skill databricks-performance-tuningコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
Zipをダウンロード ダウンロード中... name databricks-performance-tuning description Optimize Databricks cluster and query performance.
Use when jobs are running slowly, optimizing Spark configurations,
or improving Delta Lake query performance.
Trigger with phrases like "databricks performance", "spark tuning",
"databricks slow", "optimize databricks", "cluster performance".
allowed-tools Read, Write, Edit, Bash(databricks:*), Grep version 1.0.0 license MIT author Jeremy Longshore <jeremy@intentsolutions.io>
Databricks Performance Tuning
Overview
Optimize Databricks cluster, Spark, and Delta Lake performance.
Prerequisites
Access to cluster configuration
Understanding of workload characteristics
Query history access
Instructions
Step 1: Cluster Sizing
def recommend_cluster_size
data_size_gb: float ,
complexity: str = "medium" ,
parallelism_need: str = "standard" ,
dict
"""
Recommend cluster configuration based on workload.
Args:
data_size_gb: Estimated data size to process
complexity: Query/transform complexity
parallelism_need: Required parallelism level
Returns:
Recommended cluster configuration
"""
14
max
1
int
2
"low"
1
"medium"
1.5
"high"
2.5
int
1.5
if
"high"
max
8
return
"node_type_id"
"Standard_DS3_v2"
"num_workers"
"autoscale"
"min_workers"
max
1
2
"max_workers"
2
"spark_conf"
"spark.sql.shuffle.partitions"
str
4
"spark.default.parallelism"
str
4
Step 2: Spark Configuration Optimization
spark_configs = {
"etl_batch" : {
"spark.sql.shuffle.partitions" : "200" ,
"spark.default.parallelism" : "200" ,
"spark.sql.files.maxPartitionBytes" : "134217728" ,
"spark.databricks.delta.optimizeWrite.enabled" : "true" ,
"spark.databricks.delta.autoCompact.enabled" : "true" ,
"spark.databricks.delta.properties.defaults.autoOptimize.optimizeWrite" : "true" ,
"spark.sql.adaptive.enabled" : "true" ,
"spark.sql.adaptive.coalescePartitions.enabled" : "true" ,
"spark.sql.adaptive.skewJoin.enabled" : "true" ,
},
"ml_training" : {
"spark.driver.memory" : "16g" ,
"spark.executor.memory" : "16g" ,
"spark.memory.fraction" : "0.8" ,
"spark.memory.storageFraction" : "0.3" ,
"spark.serializer" : "org.apache.spark.serializer.KryoSerializer" ,
"spark.kryoserializer.buffer.max" : "1024m" ,
},
"streaming" : {
"spark.sql.streaming.schemaInference" : "true" ,
"spark.sql.streaming.checkpointLocation" : "/mnt/checkpoints" ,
"spark.databricks.delta.autoCompact.minNumFiles" : "10" ,
"spark.sql.streaming.forEachBatch.enabled" : "true" ,
},
"interactive" : {
"spark.databricks.cluster.profile" : "singleNode" ,
"spark.master" : "local[*]" ,
"spark.sql.inMemoryColumnarStorage.compressed" : "true" ,
"spark.sql.inMemoryColumnarStorage.batchSize" : "10000" ,
}
}
Step 3: Delta Lake Optimization
from pyspark.sql import SparkSession
def optimize_delta_table (
spark: SparkSession,
table_name: str ,
z_order_columns: list [str ] = None ,
vacuum_hours: int = 168 ,
) -> dict :
"""
Optimize Delta table for query performance.
Args:
spark: SparkSession
table_name: Fully qualified table name
z_order_columns: Columns for Z-ordering (max 4)
vacuum_hours: Retention period for vacuum
Returns:
Optimization results
"""
results = {}
if z_order_columns:
z_order_clause = ", " .join(z_order_columns[:4 ])
spark.sql(f"OPTIMIZE {table_name} ZORDER BY ({z_order_clause} )" )
results["z_order" ] = z_order_columns
else :
spark.sql(f"OPTIMIZE {table_name} " )
results["optimized" ] = True
spark.sql(f"ANALYZE TABLE {table_name} COMPUTE STATISTICS" )
results["statistics_computed" ] = True
spark.sql(f"VACUUM {table_name} RETAIN {vacuum_hours} HOURS" )
results["vacuumed" ] = True
detail = spark.sql(f"DESCRIBE DETAIL {table_name} " ).first()
results["metrics" ] = {
"num_files" : detail.numFiles,
"size_bytes" : detail.sizeInBytes,
"partitions" : detail.partitionColumns,
}
return results
def enable_liquid_clustering (
spark: SparkSession,
table_name: str ,
cluster_columns: list [str ],
) -> None :
"""
Enable Liquid Clustering for automatic data layout optimization.
Liquid Clustering replaces traditional partitioning and Z-ordering
with automatic, incremental clustering.
"""
columns = ", " .join(cluster_columns)
spark.sql(f"""
ALTER TABLE {table_name}
CLUSTER BY ({columns} )
""" )
def enable_predictive_optimization (
spark: SparkSession,
table_name: str ,
) -> None :
"""Enable Databricks Predictive Optimization."""
spark.sql(f"""
ALTER TABLE {table_name}
SET TBLPROPERTIES (
'delta.enableDeletionVectors' = 'true',
'delta.targetFileSize' = '104857600'
)
""" )
Step 4: Query Performance Analysis
SELECT
query_id,
query_text,
duration / 1000 as seconds,
rows_produced,
bytes_read,
start_time
FROM system.query.history
WHERE duration > 60000
AND start_time > current_timestamp () - INTERVAL 24 HOURS
ORDER BY duration DESC
LIMIT 20 ;
EXPLAIN FORMATTED
SELECT * FROM main.silver.orders
WHERE order_date > '2024-01-01'
AND region = 'US' ;
SELECT
table_name,
SUM (bytes_read) / 1024 / 1024 / 1024 as gb_read,
SUM (rows_produced) as total_rows,
COUNT (* ) as query_count
FROM system.query.history
WHERE start_time > current_timestamp () - INTERVAL 7 DAYS
GROUP BY table_name
ORDER BY gb_read DESC ;
Step 5: Caching Strategy
from pyspark.sql import DataFrame
from functools import lru_cache
class CacheManager :
"""Manage Spark DataFrame caching."""
def __init__ (self, spark: SparkSession ):
self .spark = spark
self ._cache_registry = {}
def cache_table (
self,
table_name: str ,
cache_level: str = "MEMORY_AND_DISK" ,
) -> DataFrame:
"""Cache table with specified storage level."""
if table_name in self ._cache_registry:
return self ._cache_registry[table_name]
df = self .spark.table(table_name)
if cache_level == "MEMORY_ONLY" :
df.cache()
elif cache_level == "MEMORY_AND_DISK" :
from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)
elif cache_level == "DISK_ONLY" :
from pyspark import StorageLevel
df.persist(StorageLevel.DISK_ONLY)
df.count()
self ._cache_registry[table_name] = df
return df
def uncache_all (self ):
"""Clear all cached DataFrames."""
for df in self ._cache_registry.values():
df.unpersist()
self ._cache_registry.clear()
self .spark.catalog.clearCache()
Step 6: Join Optimization from pyspark.sql import DataFrame
from pyspark.sql.functions import broadcast
def optimize_join (
df_large: DataFrame,
df_small: DataFrame,
join_key: str ,
small_table_threshold_mb: int = 100 ,
) -> DataFrame:
"""
Optimize join based on table sizes.
Uses broadcast join for small tables,
sort-merge join for large tables.
"""
small_size_mb = df_small.count() * 100 / 1024 / 1024
if small_size_mb < small_table_threshold_mb:
return df_large.join(broadcast(df_small), join_key)
else :
return df_large.join(df_small, join_key, "inner" )
def create_bucketed_table (
spark: SparkSession,
df: DataFrame,
table_name: str ,
bucket_columns: list [str ],
num_buckets: int = 100 ,
):
"""Create bucketed table for join optimization."""
(
df.write
.bucketBy(num_buckets, *bucket_columns)
.sortBy(*bucket_columns)
.saveAsTable(table_name)
)
Output
Optimized cluster configuration
Tuned Spark settings
Optimized Delta tables
Improved query performance
Error Handling Issue Cause Solution OOM errors Insufficient memory Increase executor memory or reduce partition size Skewed data Uneven distribution Use salting or AQE skew handling Slow joins Large shuffle Use broadcast for small tables Too many files Small files problem Run OPTIMIZE regularly
Examples
Performance Benchmark import time
def benchmark_query (spark, query: str , runs: int = 3 ) -> dict :
"""Benchmark query execution time."""
times = []
for _ in range (runs):
spark.catalog.clearCache()
start = time.time()
spark.sql(query).collect()
times.append(time.time() - start)
return {
"min" : min (times),
"max" : max (times),
"avg" : sum (times) / len (times),
"runs" : runs,
}
Resources
Next Steps For cost optimization, see databricks-cost-tuning.
