Serverless Model Serving: Pay-Per-Request ML Inference

Serverless model serving eliminates infrastructure management while providing cost-effective, scalable ML inference. This guide covers implementing serverless patterns in Databricks.

Serverless Benefits

SERVERLESS_ADVANTAGES = {
    "cost_efficiency": {
        "description": "Pay only for actual inference requests",
        "best_for": ["Variable workloads", "Dev/test environments", "Batch-heavy patterns"]
    },
    "scalability": {
        "description": "Automatic scaling from zero to high throughput",
        "best_for": ["Unpredictable traffic", "Burst handling", "Global deployments"]
    },
    "simplicity": {
        "description": "No infrastructure management",
        "best_for": ["Small teams", "Rapid prototyping", "Focus on ML not ops"]
    }
}

Creating Serverless Endpoints

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput,
    ServedEntityInput
)

client = WorkspaceClient()

def create_serverless_endpoint(
    name: str,
    model_name: str,
    model_version: str,
    min_scale: int = 0,
    max_scale: int = 10
):
    """Create a serverless model endpoint"""

    endpoint = client.serving_endpoints.create_and_wait(
        name=name,
        config=EndpointCoreConfigInput(
            served_entities=[
                ServedEntityInput(
                    entity_name=model_name,
                    entity_version=model_version,
                    # Serverless configuration
                    workload_size="Small",
                    scale_to_zero_enabled=(min_scale == 0),
                    # Custom scaling bounds
                    min_instances=min_scale,
                    max_instances=max_scale
                )
            ]
        )
    )

    print(f"Endpoint created: {name}")
    print(f"State: {endpoint.state.ready}")

    return endpoint

# Create endpoint that scales to zero
endpoint = create_serverless_endpoint(
    name="recommendation-engine",
    model_name="main.ml_models.product_recommender",
    model_version="1",
    min_scale=0,
    max_scale=5
)

Cold Start Optimization

import time
import requests
from typing import Optional

class ServerlessEndpointClient:
    """Client optimized for serverless endpoints"""

    def __init__(
        self,
        workspace_url: str,
        token: str,
        warm_up_enabled: bool = True
    ):
        self.workspace_url = workspace_url
        self.token = token
        self.warm_up_enabled = warm_up_enabled
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def predict(
        self,
        endpoint_name: str,
        inputs: list,
        timeout: int = 60
    ) -> dict:
        """Make prediction with cold start handling"""

        url = f"{self.workspace_url}/serving-endpoints/{endpoint_name}/invocations"

        # Retry logic for cold starts
        max_retries = 3
        retry_delay = 5

        for attempt in range(max_retries):
            try:
                response = requests.post(
                    url,
                    headers=self.headers,
                    json={"inputs": inputs},
                    timeout=timeout
                )

                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 503:
                    # Endpoint scaling up
                    print(f"Endpoint scaling up, retry in {retry_delay}s...")
                    time.sleep(retry_delay)
                else:
                    response.raise_for_status()

            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    print(f"Request timed out, retry {attempt + 1}/{max_retries}")
                    time.sleep(retry_delay)
                else:
                    raise

        raise Exception("Max retries exceeded")

    def warm_up(self, endpoint_name: str, sample_input: dict):
        """Send warm-up request to prevent cold start"""

        if not self.warm_up_enabled:
            return

        try:
            self.predict(endpoint_name, [sample_input], timeout=120)
            print(f"Endpoint {endpoint_name} warmed up")
        except Exception as e:
            print(f"Warm-up failed (non-critical): {e}")

    def batch_predict(
        self,
        endpoint_name: str,
        inputs: list,
        batch_size: int = 100
    ) -> list:
        """Batch predictions with optimal batching"""

        results = []

        for i in range(0, len(inputs), batch_size):
            batch = inputs[i:i + batch_size]
            result = self.predict(endpoint_name, batch)
            results.extend(result.get("predictions", []))

        return results

# Usage
client = ServerlessEndpointClient(
    workspace_url="https://adb-xxx.azuredatabricks.net",
    token="your-token"
)

# Warm up endpoint (e.g., before expected traffic)
client.warm_up(
    "recommendation-engine",
    sample_input={"user_id": "test", "product_id": "test"}
)

# Make predictions
predictions = client.predict(
    "recommendation-engine",
    inputs=[{"user_id": "user123", "product_id": "prod456"}]
)

Cost Optimization Strategies

class CostOptimizedServing:
    """Cost optimization patterns for serverless serving"""

    def __init__(self, workspace_client):
        self.client = workspace_client

    def get_usage_report(self, endpoint_name: str, days: int = 30) -> dict:
        """Generate usage and cost report"""

        # Query usage metrics
        usage = spark.sql(f"""
            SELECT
                DATE(timestamp) as date,
                COUNT(*) as request_count,
                SUM(compute_duration_ms) / 1000 / 60 as compute_minutes,
                AVG(input_tokens + output_tokens) as avg_tokens
            FROM main.ml_monitoring.{endpoint_name.replace('-', '_')}_usage
            WHERE timestamp >= current_date() - INTERVAL {days} DAYS
            GROUP BY DATE(timestamp)
            ORDER BY date
        """).collect()

        # Calculate costs (simplified)
        total_compute_minutes = sum(row["compute_minutes"] for row in usage)
        total_requests = sum(row["request_count"] for row in usage)

        # Cost estimates (example rates)
        cost_per_minute = 0.01
        estimated_cost = total_compute_minutes * cost_per_minute

        return {
            "total_requests": total_requests,
            "total_compute_minutes": total_compute_minutes,
            "estimated_cost": estimated_cost,
            "cost_per_request": estimated_cost / total_requests if total_requests > 0 else 0,
            "daily_breakdown": usage
        }

    def recommend_configuration(self, usage_report: dict) -> dict:
        """Recommend endpoint configuration based on usage"""

        avg_daily_requests = usage_report["total_requests"] / 30

        if avg_daily_requests < 100:
            recommendation = {
                "config": "serverless_scale_to_zero",
                "workload_size": "Small",
                "min_instances": 0,
                "reason": "Low traffic - maximize cost savings with scale to zero"
            }
        elif avg_daily_requests < 10000:
            recommendation = {
                "config": "serverless_warm",
                "workload_size": "Small",
                "min_instances": 1,
                "reason": "Moderate traffic - keep one instance warm to avoid cold starts"
            }
        else:
            recommendation = {
                "config": "provisioned",
                "workload_size": "Medium",
                "min_instances": 2,
                "reason": "High traffic - consider provisioned throughput for cost predictability"
            }

        return recommendation

    def implement_caching(self, endpoint_name: str, cache_ttl: int = 300):
        """Implement response caching to reduce inference calls"""

        # Cache configuration
        return {
            "strategy": "lru_cache",
            "max_size": 10000,
            "ttl_seconds": cache_ttl,
            "key_function": "hash(input_features)"
        }

# Usage
optimizer = CostOptimizedServing(client)

# Get usage report
report = optimizer.get_usage_report("recommendation-engine", days=30)
print(f"Total cost: ${report['estimated_cost']:.2f}")
print(f"Cost per request: ${report['cost_per_request']:.4f}")

# Get recommendations
recommendation = optimizer.recommend_configuration(report)
print(f"Recommended config: {recommendation['config']}")
print(f"Reason: {recommendation['reason']}")

Hybrid Serving Patterns

class HybridServingRouter:
    """Route between serverless and provisioned endpoints"""

    def __init__(self, serverless_endpoint: str, provisioned_endpoint: str):
        self.serverless = serverless_endpoint
        self.provisioned = provisioned_endpoint
        self.request_count = 0
        self.threshold = 100  # Requests per minute threshold

    def route(self, inputs: list, latency_requirement: str = "normal") -> str:
        """Determine which endpoint to use"""

        if latency_requirement == "low":
            # Use provisioned for low-latency requirements
            return self.provisioned
        elif self.request_count > self.threshold:
            # High traffic - use provisioned for cost efficiency
            return self.provisioned
        else:
            # Normal traffic - use serverless
            return self.serverless

    def predict(
        self,
        inputs: list,
        latency_requirement: str = "normal"
    ) -> dict:
        """Make prediction with automatic routing"""

        endpoint = self.route(inputs, latency_requirement)

        # Make prediction
        result = self._call_endpoint(endpoint, inputs)

        # Track request for routing decisions
        self.request_count += 1

        return {
            "predictions": result,
            "endpoint_used": endpoint
        }

    def _call_endpoint(self, endpoint: str, inputs: list) -> dict:
        # Implementation
        pass

# Usage
router = HybridServingRouter(
    serverless_endpoint="model-serverless",
    provisioned_endpoint="model-provisioned"
)

# Normal request - may use serverless
result = router.predict(inputs, latency_requirement="normal")

# Low-latency request - uses provisioned
result = router.predict(inputs, latency_requirement="low")

Monitoring Serverless Endpoints

class ServerlessMonitor:
    """Monitor serverless endpoint health and costs"""

    def __init__(self, endpoint_name: str):
        self.endpoint_name = endpoint_name

    def get_scaling_metrics(self, hours: int = 24) -> dict:
        """Get scaling behavior metrics"""

        metrics = spark.sql(f"""
            SELECT
                DATE_TRUNC('hour', timestamp) as hour,
                COUNT(*) as requests,
                MAX(active_instances) as max_instances,
                MIN(active_instances) as min_instances,
                SUM(CASE WHEN cold_start THEN 1 ELSE 0 END) as cold_starts
            FROM main.ml_monitoring.{self.endpoint_name.replace('-', '_')}_scaling
            WHERE timestamp >= current_timestamp() - INTERVAL {hours} HOURS
            GROUP BY DATE_TRUNC('hour', timestamp)
            ORDER BY hour
        """)

        return metrics.toPandas().to_dict('records')

    def get_cold_start_analysis(self) -> dict:
        """Analyze cold start patterns"""

        analysis = spark.sql(f"""
            SELECT
                HOUR(timestamp) as hour_of_day,
                COUNT(*) as total_requests,
                SUM(CASE WHEN cold_start THEN 1 ELSE 0 END) as cold_starts,
                AVG(CASE WHEN cold_start THEN latency_ms ELSE NULL END) as avg_cold_start_latency,
                AVG(CASE WHEN NOT cold_start THEN latency_ms ELSE NULL END) as avg_warm_latency
            FROM main.ml_monitoring.{self.endpoint_name.replace('-', '_')}_requests
            WHERE timestamp >= current_date() - INTERVAL 7 DAYS
            GROUP BY HOUR(timestamp)
            ORDER BY hour_of_day
        """).collect()

        return {
            "by_hour": analysis,
            "recommendation": self._get_warm_up_recommendation(analysis)
        }

    def _get_warm_up_recommendation(self, analysis) -> str:
        """Recommend warm-up schedule based on cold start patterns"""

        high_cold_start_hours = [
            row["hour_of_day"] for row in analysis
            if row["cold_starts"] / row["total_requests"] > 0.1
        ]

        if high_cold_start_hours:
            return f"Consider warm-up requests at hours: {high_cold_start_hours}"
        return "Cold starts are minimal, no warm-up needed"

# Usage
monitor = ServerlessMonitor("recommendation-engine")
cold_start_analysis = monitor.get_cold_start_analysis()
print(cold_start_analysis["recommendation"])

Conclusion

Serverless model serving provides cost-effective, scalable ML inference. Handle cold starts with warm-up strategies, optimize costs with caching and intelligent routing, and monitor scaling behavior for continuous improvement.