Complete Google Cloud Run for ML Tutorial: Serverless ML Deployment

Google Cloud Run provides a serverless platform for deploying containerized ML models. It offers auto-scaling, pay-per-use pricing, and seamless integration with Google Cloud services.

Why Cloud Run for ML?

• Serverless: No infrastructure management
• Auto-scaling: Scale to zero and up automatically
• Cost-effective: Pay only for actual usage
• Container-based: Deploy any framework
• Fast deployment: Deploy in seconds

Prerequisites

pip install google-cloud-run flask gunicorn

gcloud auth login
gcloud config set project your-project-id

Quick Start

1. Create ML Service

# app.py
from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(name)

Load model on startup
model = joblib.load("model.joblib")

@app.route("/predict", methods=["POST"])
def predict():
    data = request.getjson()

    features = np.array(data["features"]).reshape(1, -1)
    prediction = model.predict(features)
    probability = model.predictproba(features)

    return jsonify({
        "prediction": int(prediction[0]),
        "probability": probability[0].tolist()
    })

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "healthy"})

if name == "main":
    app.run(host="0.0.0.0", port=8080)

2. Create Dockerfile

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY model.joblib .
COPY app.py .

ENV PORT=8080
EXPOSE 8080

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 app:app

3. requirements.txt

flask==2.3.0
gunicorn==21.2.0
joblib==1.3.0
scikit-learn==1.3.0
numpy==1.24.0

4. Deploy to Cloud Run

# Build container
gcloud builds submit --tag gcr.io/your-project/ml-service

Deploy
gcloud run deploy ml-service \
  --image gcr.io/your-project/ml-service \
  --platform managed \
  --region us-central1 \
  --memory 2Gi \
  --cpu 2 \
  --min-instances 0 \
  --max-instances 10 \
  --allow-unauthenticated

FastAPI Service

1. FastAPI Application

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI(title="ML Prediction API")

Load model
model = joblib.load("model.joblib")

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: int
    probability: list[float]

@app.post("/predict", responsemodel=PredictionResponse)

async def predict(request: PredictionRequest):
    try:
        features = np.array(request.features).reshape(1, -1)
        prediction = model.predict(features)
        probability = model.predictproba(features)

        return PredictionResponse(
            prediction=int(prediction[0]),
            probability=probability[0].tolist()
        )
    except Exception as e:
        raise HTTPException(statuscode=400, detail=str(e))


@app.get("/health")
async def health():
    return {"status": "healthy"}

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY model.joblib .
COPY main.py .

ENV PORT=8080
EXPOSE 8080

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

# servepytorch.py
from flask import Flask, request, jsonify
import torch
import torch.nn as nn
import numpy as np

app = Flask(name)

class SimpleNN(nn.Module):
    def init(self, inputsize, hiddensize, outputsize):

        super().init()
        self.fc1 = nn.Linear(inputsize, hiddensize)

        self.fc2 = nn.Linear(hiddensize, outputsize)

        self.relu = nn.ReLU()
        self.softmax = nn.Softmax(dim=1)

    def forward(self, x):
        x = self.relu(self.fc1(x))
        x = self.softmax(self.fc2(x))
        return x

Load model
model = SimpleNN(10, 50, 2)
model.loadstatedict(torch.load("model.pt", maplocation="cpu"))
model.eval()

@app.route("/predict", methods=["POST"])
def predict():
    data = request.getjson()

    features = torch.tensor(data["features"], dtype=torch.float32)

    with torch.nograd():
        output = model(features)
        prediction = torch.argmax(output, dim=1)

    return jsonify({
        "prediction": prediction.tolist(),
        "probabilities": output.tolist()
    })

if name == "main":
    app.run(host="0.0.0.0", port=8080)

TensorFlow Serving

1. TensorFlow Service

# servetensorflow.py
from flask import Flask, request, jsonify
import tensorflow as tf
import numpy as np

app = Flask(name)

Load SavedModel
model = tf.savedmodel.load("savedmodel")


@app.route("/predict", methods=["POST"])
def predict():
    data = request.getjson()
    features = np.array(data["features"], dtype=np.float32)

    predictions = model(features)

    return jsonify({
        "predictions": predictions.numpy().tolist()
    })

if name == "main":
    app.run(host="0.0.0.0", port=8080)

Batch Processing

1. Batch Prediction Endpoint

from flask import Flask, request, jsonify
import joblib
import numpy as np
from concurrent.futures import ThreadPoolExecutor

app = Flask(name)
model = joblib.load("model.joblib")
executor = ThreadPoolExecutor(maxworkers=4)


def predictbatch(batch):
    return model.predict(batch)

@app.route("/predict/batch", methods=["POST"])
def batchpredict():

    data = request.getjson()
    features = np.array(data["features"])

    # Process in parallel
    batchsize = 100

    batches = [features[i:i+batchsize] for i in range(0, len(features), batchsize)]


    futures = [executor.submit(predictbatch, batch) for batch in batches]
    results = [f.result() for f in futures]

    predictions = np.concatenate(results)

    return jsonify({
        "predictions": predictions.tolist()
    })

Model Versioning

1. Multiple Model Versions

# multimodelservice.py
from flask import Flask, request, jsonify
import joblib
import os

app = Flask(name)

Load multiple model versions
models = {}
modeldir = "models/"

for version in os.listdir(modeldir):
    models[version] = joblib.load(f"{modeldir}/{version}/model.joblib")


@app.route("/predict/", methods=["POST"])
def predict(version):
    if version not in models:
        return jsonify({"error": "Model version not found"}), 404

    data = request.getjson()
    model = models[version]
    prediction = model.predict([data["features"]])

    return jsonify({
        "version": version,
        "prediction": int(prediction[0])
    })

Authentication

1. IAM Authentication

# Deploy with authentication required
gcloud run deploy ml-service \
  --image gcr.io/your-project/ml-service \
  --no-allow-unauthenticated

Grant access
gcloud run services add-iam-policy-binding ml-service \
  --member="serviceAccount:caller@project.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

2. Call Authenticated Service

import google.auth.transport.requests
import google.oauth2.idtoken

import requests

def callauthenticatedservice(url, data):

    authreq = google.auth.transport.requests.Request()
    idtoken = google.oauth2.idtoken.fetchidtoken(authreq, url)


    response = requests.post(
        f"{url}/predict",
        json=data,
        headers={"Authorization": f"Bearer {idtoken}"}
    )
    return response.json()

CI/CD with Cloud Build

1. cloudbuild.yaml

steps:
  # Build container
name: 'gcr.io/cloud-builders/docker'
    args: ['build', '-t', 'gcr.io/$PROJECTID/ml-service:$COMMITSHA', '.']


  # Push to registry
name: 'gcr.io/cloud-builders/docker'
    args: ['push', 'gcr.io/$PROJECTID/ml-service:$COMMITSHA']


  # Deploy to Cloud Run
name: 'gcr.io/google.com/cloudsdktool/cloud-sdk'
    entrypoint: gcloud

    args:
'run'
'deploy'
'ml-service'
'--image'
'gcr.io/$PROJECTID/ml-service:$COMMITSHA'
'--region'
'us-central1'
'--platform'
'managed'

images:
'gcr.io/$PROJECTID/ml-service:$COMMITSHA'

Monitoring

1. Custom Metrics

from flask import Flask, request, jsonify
from google.cloud import monitoringv3

import time

app = Flask(name)
client = monitoringv3.MetricServiceClient()
projectname = f"projects/your-project"


def recordlatency(latencyms):

    series = monitoringv3.TimeSeries()
    series.metric.type = "custom.googleapis.com/ml/predictionlatency"

    series.resource.type = "global"

    point = monitoringv3.Point()
    point.value.doublevalue = latencyms
    point.interval.endtime.seconds = int(time.time())

    series.points = [point]

    client.createtimeseries(name=projectname, timeseries=[series])


@app.route("/predict", methods=["POST"])
def predict():
    start = time.time()

    # Prediction logic
    result = model.predict(...)

    latency = (time.time() - start) * 1000
    recordlatency(latency)

    return jsonify({"prediction": result})

Best Practices

1. Optimize Container

# Use multi-stage build
FROM python:3.9-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt

FROM python:3.9-slim
COPY --from=builder /root/.local /root/.local
COPY . .

ENV PATH=/root/.local/bin:$PATH
CMD ["gunicorn", "--bind", ":8080", "app:app"]

2. Warm-up Requests

@app.beforefirstrequest
def warmup():
    # Make dummy prediction to warm up model
    dummyinput = np.zeros((1, 10))

    model.predict(dummyinput)
    print("Model warmed up")

Conclusion

Cloud Run for ML provides:

Key takeaways:

• Containerize your ML models
• Optimize for cold starts
• Use appropriate memory/CPU
• Implement health checks
• Monitor performance metrics