Optimise your Databricks Costs

This article describes all the costs associated with executing workloads on Databricks and provides a structured framework for optimising your cloud spend.

Databricks Workload Cost Components

First let’s start by understanding the Databricks pricing model. Workloads running on Databricks incur Databricks costs (DBU) and cloud infrastructure costs.

Databricks Costs

Databricks offers you a pay-as-you-go approach with no up-front, license, startup costs or a monthly fee. One only pays when clusters/compute is running at per second granularity. Workspace and Unity Catalog with no compute running incurs no charges.

Databricks charges for compute usage in Databricks Units (DBUs). The number of DBUs a workload consumer depends on

• Workspace Tier viz. Standard,? Premium or Enterprise
• Cluster type (Job, All purpose, DLT, SQL Warehouse, Serverless, etc.)
• Cluster/warehouse size
• Photon acceleration
• Time the compute instances are running. These are billed per second.

Cloud Infrastructure Costs

These include.

1. Cloud Compute (exception VM costs are integrated in the DBU costs in the Databricks serverless offering)

• VMs
• VM disks
• Networking

2. Cloud Storage (ADLS, S3, GCS) Managed Databricks Storage (DBFS)

3. Custom cloud services (optional)

• Private Endpoints
• Firewalls
• Diagnostic/Audit Logs delivery cost

Cost Optimisation Levers

Now that you understanding the Databricks charging, let us move on to cost optimisation. There are 5 components that drive the costs of a your Databricks workload.

1.??Model (Map and Reduce)

2. (Efficient) Distribution of your code on Spark

2.??Storage organisation

3.??Cloud Infrastructure options. These drive the cloud resource costs (compute, storage, network, etc.) used by your Spark jobs.

4.?Databricks Cluster configuration. These drive the (DBU) costs.

The diagram below shows the further drivers that form part of these components.

Cost Optimisation Strategy (Impact vs Effort)

The diagram classifies the recommendations by impact and effort. Often for existing jobs there is a limited degree of freedom on how much code changes can be made, hence I would recommend the following approach

• For existing workloads Look at quick win items first and put high impact items on the backlog
• For brand new jobs use a holistic approach. Ensure that you cover the entire checklist and tackle the high impact items first.

1. Model Implementation

Efficient Map and Reduce Operations

• Spark is a distributed computing framework that allows calculations for a single row/entity to be run in parallel over your entire dataset.
• You need to ensure that your single row/entity calculation is efficient.
• Spark is not going to help to distribute an inefficient implementation of your model.
• Ensure your map and reduce functions are efficient and unit tested before distributing them using Spark. It is far easier and cheaper to develop, debug and test code locally rather than on a distributed cluster.

Spark Version

Use the latest version of Spark where possible (preferably one with Long Term Support (LTS) from Databricks. e.g.?Spark 3.2.1 (DBR 10.4 LTS) is 10% faster than Spark 2.4.5 without any other changes

2. Job Distribution

One of the big drivers of costs is how efficiently and evenly is your model distributed on the Spark cluster. The 5Sses usually are the most common causes of performance and hence costs.

• Serialization - The distribution of data and code segments across the cluster.
• Shuffles - The act of moving data between executors
• Skew - imbalance in partition sizes
• Spill - The writing of temp files to disk due to lack of memory.
• Storage - efficiently reading data from storage

2.1 Serialisation

The Spark DataFrame representation and instructions are compact, and optimized for distributing from the Driver to each Executor. Avoid using custom code when there is an existing built-in Dataframe function that can accomplish the same task. DataFrame and built in Spark operators are the fastest, use these where possible over RDDs and custom UDFs.

Dataframes are stored using Spark's Tungsten's internal memory representation using Encoders. As a result, Tungsten Encoders can efficiently serialize/deserialize JVM objects as well as generate compact bytecode that can execute at superior speeds.

Spark uses the Catalyst Optimizer to generate an optimized logical and physical query plan. RDDs and custom UDFs create an analysis barrier for the Catalyst Optimizer. The Catalyst Optimizer cannot connect code before and after UDF. The UDF is a black box which means it limits optimizations to the code before and after, excluding the UDF and how all the code works together.

User Defined Functions

If you have to use them …

? Use Vectorized UDFs over “standard” Python UDFs

? Use Typed Transformations over “standard” Scala UDFs

2.2 Shuffles

Shuffling is a side effect of wide transformation like join(), distinct(), groupBy() and orderBy(). Shuffling data between nodes in a cluster can be time-consuming and resource-intensive. You can mitigate shuffles by

• Minimize data shuffling by narrowing columns, preemptively filtering data and designing Spark jobs that make efficient use of the data already present in each node. Use partitioning and the right selection of keys.
• Broadcast small tables
• Increase the spark.sql.autoBroadcastJoinThreshold parameter. The default setting is too low in most cases.
• Reduce network IO by using fewer and larger workers
• Consider denormalising your schema. Delta Lake storage is generally cheap. Trade off storage vs compute
• Optimise shuffles by choosing the right shuffle strategy, no of partitions, using AQE

2.3 Skew

• Data in Spark typically read in as 128MB partitions and evenly distributed
• Transformations and aggregations may cause some partitions to have significantly more records in one partition than another
• Large skews can result in low utilization, spill or Out of Memory (OOM) errors.

Skew can be mitigated by using AQE. AQE automatically optimises Skewed joins by splitting skewed tasks/partitions into roughly even sized ones.

2.4 Spill

Spill is the term used to refer to the act of moving an RDD from RAM to disk and later back into RAM again. This occurs when the given partition is simply too big to fit into RAM. Spilling requires expensive disk writes and read to free up local RAM

Spill can occur when

• Setting spark.sql.files.maxPartitionBytes to high (default is 128 MB)
• Highly compressed datasets
• The explode() of even a small array
• The join() or crossJoin() of two tables
• Aggregating results by a skewed feature

You can mitigate spill by

• Assigning memory per worker by using memory intensive worker types
• Using AQE. AQE splits large partitions into smaller chunks.
• Decrease partition size by managing spark.sql.shuffle.partition, explicitly repartitioning or configuring spark.sql.files.maxPartitionBytes

2.6 Caching

Cache and persist data: Caching data in memory can significantly speed up Spark jobs. When performing iterative or multiple computations on the same data, persisting it in memory can improve performance.

3.0 Delta Storage Optimisation

For in-depth guidance on storage optimization, refer to my article on Delta Lake optimisation best practices. Here's a summary:

• Embrace the Delta format: Leverage Delta's capabilities for efficient storage management.
• Optimize schema design: Choose appropriate data types and denormalization levels.
• Prioritize filtering and statistics: Order frequently filtered columns first and gather statistics for query optimization.
• Minimize read IO: Implement strategies like partitioning, Z-ordering, and Bloom filters to skip irrelevant data during reads.
• Maintain optimal file sizes: Regularly run and commands and monitor your lakehouse for optimal performance.
• Utilize Delta Cache: Cache frequently used datasets for faster access.

4.0 Cloud Infrastructure Cost Optimisation

Cloud costs are primarily driven by VM compute costs. They key is to right-size the driver and worker instance types, using spot instances when appropriate. Driver and workers perform very different functions in Spark and must be sized independently. Secondly locate your Databricks workspace in the right region to minimize network bandwidth (colocated with data) and compute costs.

4.1 Spark Driver Node Sizing

Always use an on-demand instance for drivers. Start with 4 cores and 16GB RAM, unless

• Large data sets are being returned/collected (spark.driver.maxResultSize is typically increased also)
• Large dataframes are being broadcast joins
• It is a shared cluster with many concurrent jobs/notebooks

4.2 Common VM Instance Types

Familiarize yourself with the VM instance types offered by your cloud provider (e.g., Azure instance types and VM instance types). Here's a breakdown of common types:

• General-purpose: A good starting point for mixed workloads.

• Memory-optimized: Ideal for data caching, large shuffles/spills, and spark caching.
• Compute-optimized: Suitable for ETL workloads with full file scans and no data reuse.
• Storage-optimized: Optimized for Delta IO caching, especially for workloads with persistent shuffle-spills.
• GPU-optimized: Designed for machine learning and deep learning workloads with high memory and compute demands.

4.3 Selecting the Right Instance Type and Size

• Start with general-purpose instances: Use these as a baseline.
• Memory-optimized instances: Choose these for workloads with significant spills or large working sets.
• Compute-optimized instances: Opt for these for compute-intensive workloads such as ETL jobs without shuffles.
• Storage-optimized instances: Use these for workloads with large shuffles.
• Delta cache-accelerated instances: Select these if your working set you're using Delta Cache. They offer faster local SSD storage.
• Instance size: Generally, a smaller number of larger machines is preferable (unless resilience is critical) to minimize network IO between nodes. This is especially true for shuffle-heavy workloads.
• RAM limitations: Avoid worker nodes with more than 128GB of RAM to prevent long garbage collection pauses. Consider using Delta Cache instead of Spark Cache, minimizing operations, and utilizing G1 garbage collection to mitigate this.
• Processor generation: Choose the latest generation processors for better price-performance (e.g., Ev4/Edv4 over Ev2/Edv3).
• AMD vs. Intel: AMD processors are often more cost-effective (~10-15%) than their Intel counterparts and run workloads seamlessly due to x86 instruction set compatibility.

4.4 Region Selection

Select the Databricks workspace region with lowest compute costs one one that minimizes network bandwidth costs by co-locating it with your data.

4.5 Storage Lifecycle Policies

• Choose the appropriate redundancy level: Select the storage redundancy level (LRS < ZRS < GRS < RA-GRS < GZRS < RA-GZRS, etc.) that aligns with your needs. Higher redundancy comes at a higher cost.
• Implement lifecycle management: Set up policies to move data between storage tiers (hot/cold) and delete data exceeding retention periods.
• Clean up temporary data: Regularly remove checkpoint, cluster log, and application data to prevent storage cost accumulation.
• Storage access control: Use service or private accounts for storage access, avoiding NAT gateways.

5.0 Databricks Cost Optimisation

Databricks charges for compute usage in units called Databricks Units (DBUs). The number of DBUs you consume depends on several configuration choices:

• Workspace Tier: Standard, Premium, or Enterprise tiers offer different features and pricing structures.
• Cluster Type: Job clusters, all-purpose clusters, DLT clusters, SQL warehouses, serverless clusters, etc., each have specific use cases and costs.
• Cluster/Warehouse Size: Larger clusters cost more DBUs, but may not always be necessary. Consider that if your job finishes in half the time on a double-sized cluster, the cost remains the same.
• Photon Acceleration: SQL workloads can run significantly faster (3x-10x) with Photon, and the additional cost can be offset by the performance gains.
• Instance Runtime: Databricks bills per second for compute instance usage.

Seel Cluster Configuration Best Practices for detailed best practices.

5.1 Databricks Platform Tier

Select the Databricks platform tier that aligns with your non-functional requirements, such as security or compliance needs.

5.2 Cluster Type

• All-Purpose Compute: Ideal for prototyping, development, and basic result analysis. However, all-purpose clusters cost nearly 50% more DBUs.
• Job Compute or Job Compute Light: Choose these for model execution tasks.

5.3 Cluster Mode

• Standard: Recommended for single-user clusters that can run SQL, Python, R, and Scala workloads.
• High Concurrency: Optimized for running multiple concurrent SQL, Python, and R workloads, but not Scala. Use this for scenarios with multiple concurrent users.
• Single Node: Clusters with no workers. Suitable for single-user workloads on small data volumes, using single-node libraries (e.g., Pandas) or job steps with shell and file system commands. These libraries don't leverage the distributed nature of a cluster and worker nodes are unused.

5.4 Databricks Runtime

• Use the latest runtime version, ideally one with Long-Term Support (LTS), unless there's a compelling reason not to (e.g., incompatible dependencies). Spark 3.x offers a 10% performance improvement over Spark 2.x.
• For SQL-heavy workloads, consider the Photon runtime. Remember that Photon has additional costs. Benchmark performance with and without Photon for mixed workloads. Avoid using Photon for non-dataframe workloads (e.g., RDDs).

5.5 Auto Scaling, Instance Types and Inactivity Timeout

• Enable Autoscaling: This is beneficial when workloads vary across jobs and stages and varying user numbers for shared clusters and SQL warehouses.
• Enable Inactivity Termination: Use this for interactive clusters.
• Spot Instances: Spot virtual machines are unused compute capacity available at significant discounts (up to 90%). Enable Spot instances unless you have very strict service level agreements (SLAs).

5.6 Instance pools and reserved instances

• Set up pools for frequently used instance types. This reduces cluster startup times and allows for shorter inactivity timeouts, leading to cost savings.
• Standardize on instance types and leverage reserved instances for those you use frequently. Common instance types in a pool improve utilization, leading to faster startup times and cost savings.

5.7 Cluster Policy

Define and enforce cluster policy to ensure adherence to best practices on shared development clusters.

6.0 Top Tips for Databricks Cost Optimization

• Use the latest Spark version (currently 3.x) and the Databricks Runtime with Long Term Support (LTS).
• Prioritize use of Dataframes/Spark SQL and built-in functions whenever possible.
• Enable Automatic Query Execution (AQE) and avoid manually specifying number of partitions.
• Reduce and mitigate shuffles and spills in your data processing.
• Utilize Delta Lake features like optimizations, partitions, file sizes, Z-order, statistics, and merges.
• Avoid partitioning tables smaller than 1TB and having partitions with less than 1GB of data.
• Don't use interactive clusters for automated jobs.
• Choose v4 or the latest generation and AMD compute instance types.
• Consider Photon for SQL workloads.
• Use Spot instances (with overprovisioning) for workers if your SLAs allow for some flexibility.