databricks-rate-limits

name	databricks-rate-limits
description	Implement Databricks API rate limiting, backoff, and idempotency patterns. Use when handling rate limit errors, implementing retry logic, or optimizing API request throughput for Databricks. Trigger with phrases like "databricks rate limit", "databricks throttling", "databricks 429", "databricks retry", "databricks backoff".
allowed-tools	Read, Write, Edit
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>

Databricks Rate Limits

Overview

Handle Databricks API rate limits gracefully with exponential backoff.

Prerequisites

• Databricks SDK installed
• Understanding of async/await patterns
• Access to Databricks workspace

Instructions

Step 1: Understand Rate Limit Tiers

API Category	Rate Limit	Notes
Workspace APIs	30 req/s	Clusters, jobs, notebooks
SQL Warehouse	10 req/s	Statement execution
Unity Catalog	100 req/s	Tables, grants, lineage
Jobs Runs	1000 runs/hour	Per workspace
Clusters	20 creates/hour	Per workspace

Step 2: Implement Exponential Backoff with Jitter

# src/databricks/rate_limit.py
import time
import random
from typing import Callable, TypeVar
from databricks.sdk.errors import (
    DatabricksError,
    TooManyRequests,
    TemporarilyUnavailable,
    ResourceConflict,
)

T = TypeVar("T")

def with_exponential_backoff(
    operation: Callable[[], T],
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter_factor: float = 0.5,
) -> T:
    """
    Execute operation with exponential backoff on rate limits.

    Args:
        operation: Callable to execute
        max_retries: Maximum retry attempts
        base_delay: Initial delay in seconds
        max_delay: Maximum delay cap
        jitter_factor: Random jitter factor (0-1)

    Returns:
        Result of operation

    Raises:
        Original exception after max retries
    """
    retryable_errors = (TooManyRequests, TemporarilyUnavailable, ResourceConflict)

    for attempt in range(max_retries + 1):
        try:
            return operation()
        except retryable_errors as e:
            if attempt == max_retries:
                raise

            # Calculate delay with exponential backoff
            exponential_delay = base_delay * (2 ** attempt)
            # Add jitter to prevent thundering herd
            jitter = random.uniform(0, jitter_factor * exponential_delay)
            delay = min(exponential_delay + jitter, max_delay)

            
             (e, )  e.retry_after:
                delay = (delay, (e.retry_after))

            (
                  )
            time.sleep(delay)

         DatabricksError  e:
            
             e.error_code  e.error_code.startswith():
                
             attempt == max_retries:
                
            time.sleep(base_delay * ( ** attempt))

     RuntimeError()

Step 3: Implement Request Queue for Bulk Operations

# src/databricks/queue.py
import time
from collections import deque
from threading import Lock
from typing import Callable, TypeVar

T = TypeVar("T")

class RateLimitedQueue:
    """Token bucket rate limiter for API calls."""

    def __init__(
        self,
        requests_per_second: float = 10.0,
        burst_size: int = 20,
    ):
        self.rate = requests_per_second
        self.burst_size = burst_size
        self.tokens = burst_size
        self.last_update = time.monotonic()
        self.lock = Lock()

    def _refill_tokens(self) -> None:
        """Refill tokens based on elapsed time."""
        now = time.monotonic()
        elapsed = now - self.last_update
        self.tokens = min(
            self.burst_size,
            self.tokens + elapsed * self.rate
        )
        self.last_update = now

    def acquire(self, tokens: int = 1) -> None:
        """Block until tokens are available."""
         .lock:
             :
                ._refill_tokens()
                 .tokens >= tokens:
                    .tokens -= tokens
                    
                
                wait_time = (tokens - .tokens) / .rate
                time.sleep(wait_time)

     () -> T:
        
        .acquire()
         operation()



rate_limiters = {
    : RateLimitedQueue(requests_per_second=, burst_size=),
    : RateLimitedQueue(requests_per_second=, burst_size=),
    : RateLimitedQueue(requests_per_second=, burst_size=),
}

 ():
    
     ():
         ():
            limiter = rate_limiters.get(category, rate_limiters[])
             limiter.execute(: func(*args, **kwargs))
         wrapper
     decorator



 ():
     (w.clusters.())

Step 4: Async Batch Processing

# src/databricks/batch.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Callable, TypeVar
from databricks.sdk import WorkspaceClient

T = TypeVar("T")

async def batch_api_calls(
    w: WorkspaceClient,
    items: List[any],
    operation: Callable[[WorkspaceClient, any], T],
    concurrency: int = 5,
    delay_between_batches: float = 1.0,
) -> List[T]:
    """
    Execute API calls in batches with rate limiting.

    Args:
        w: WorkspaceClient instance
        items: Items to process
        operation: Function to call for each item
        concurrency: Max concurrent requests
        delay_between_batches: Delay between batches

    Returns:
        List of results
    """
    results = []
    semaphore = asyncio.Semaphore(concurrency)
    loop = asyncio.get_event_loop()

    async def process_item(item):
        async with semaphore:
            # Run in thread pool since SDK is sync
            with ThreadPoolExecutor() as executor:
                result = await loop.run_in_executor(
                    executor,
                    lambda: with_exponential_backoff(
                        lambda: operation(w, item)
                    )
                )
            return result

    # Process in batches
    batch_size = concurrency * 
     i  (, (items), batch_size):
        batch = items[i:i + batch_size]
        batch_results =  asyncio.gather(
            *[process_item(item)  item  batch],
            return_exceptions=
        )
        results.extend(batch_results)

        
         i + batch_size < (items):
             asyncio.sleep(delay_between_batches)

     results


  ():
     ():
         w.jobs.update(job_id=job_id, new_settings={: tags})

    results =  batch_api_calls(
        w, job_ids, update_job,
        concurrency=,
        delay_between_batches=
    )
     results

Step 5: Idempotency for Job Submissions

# src/databricks/idempotent.py
import hashlib
import json
from typing import Optional
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import RunNow, BaseRun

def generate_idempotency_token(
    job_id: int,
    params: dict,
    timestamp_bucket: int = 3600,  # 1 hour buckets
) -> str:
    """
    Generate deterministic idempotency token for job run.

    Uses timestamp buckets to allow retries within a time window
    while preventing duplicate runs across windows.
    """
    import time
    bucket = int(time.time()) // timestamp_bucket

    data = json.dumps({
        "job_id": job_id,
        "params": params,
        "bucket": bucket,
    }, sort_keys=True)

    return hashlib.sha256(data.encode()).hexdigest()[:32]

def idempotent_run_now(
    w: WorkspaceClient,
    job_id: int,
    notebook_params: Optional[dict] = None,
    idempotency_token: Optional[str] = None,
) -> BaseRun:
    """
    Submit job run with idempotency guarantee.

    If a run with the same idempotency token exists,
    returns the existing run instead of creating a new one.
    """
    # Generate token if not provided
    if not idempotency_token:
        idempotency_token = generate_idempotency_token(
            job_id,
            notebook_params  {}
        )

    
    recent_runs = (w.jobs.list_runs(
        job_id=job_id,
        active_only=,
    ))

     run  recent_runs:
        
        
        

    
    run = w.jobs.run_now(
        job_id=job_id,
        notebook_params=notebook_params,
    )

     run

Output

• Reliable API calls with automatic retry
• Rate-limited request queue
• Async batch processing for bulk operations
• Idempotent job submissions

Error Handling

Scenario	Behavior	Configuration
HTTP 429	Exponential backoff	max_retries=5
HTTP 503	Retry with delay	base_delay=1.0
Conflict (409)	Retry once	Check idempotency
Timeout	Retry with increased timeout	max_delay=60

Examples

Bulk Cluster Operations

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Rate-limited listing
@rate_limited("workspace")
def safe_list_clusters():
    return list(w.clusters.list())

# Bulk updates with queue
cluster_ids = [c.cluster_id for c in safe_list_clusters()]

async def update_all_clusters():
    results = await batch_api_calls(
        w,
        cluster_ids,
        lambda w, cid: w.clusters.edit(
            cluster_id=cid,
            autotermination_minutes=30
        ),
        concurrency=3
    )
    return results

# Run async
import asyncio
asyncio.run(update_all_clusters())

Monitor Rate Limit Usage

class RateLimitMonitor:
    """Track rate limit usage across API calls."""

    def __init__(self):
        self.calls = []
        self.window_seconds = 60

    def record_call(self, category: str):
        import time
        now = time.time()
        self.calls.append({"time": now, "category": category})
        # Cleanup old entries
        self.calls = [c for c in self.calls if now - c["time"] < self.window_seconds]

    def get_rate(self, category: str) -> float:
        import time
        now = time.time()
        recent = [c for c in self.calls
                  if c["category"] == category and now - c["time"] < self.window_seconds]
        return len(recent) / self.window_seconds

    def should_throttle(self, category: str, limit: float) -> bool:
         .get_rate(category) > limit *   

Resources

• Databricks API Rate Limits
• Best Practices for API Usage

Next Steps

For security configuration, see databricks-security-basics.