Deploying ML Models on AWS, Azure and GCP

import os

import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException, Request, Response

# SageMaker copies the model to /opt/ml/model; Google passes its location in AIP_STORAGE_URI. MODEL_DIR overrides both locally.
MODEL_DIR = os.environ.get("MODEL_DIR", "/opt/ml/model")
model = joblib.load(os.path.join(MODEL_DIR, "model.joblib"))
FEATURES = list(model.feature_names_in_)
app = FastAPI()


def predict(body):
    if not isinstance(body, dict) or not isinstance(body.get("instances"), list):   # a client error, not a server fault
        raise HTTPException(status_code=400, detail='expected a JSON object with an "instances" list')
    return [round(float(v), 2) for v in model.predict(pd.DataFrame(body["instances"], columns=FEATURES))]


# Amazon SageMaker AI contract: port 8080, GET /ping, POST /invocations
@app.get("/ping")
def ping():
    return Response(status_code=200)


@app.post("/invocations")
async def invocations(request: Request):
    body = await request.json()
    return {"predictions": predict(body)}


# Google Cloud (Gemini Enterprise Agent Platform) contract: the platform sets the routes and the port
@app.get(os.environ.get("AIP_HEALTH_ROUTE", "/health"))
def health():
    return {"status": "ok"}


@app.post(os.environ.get("AIP_PREDICT_ROUTE", "/predict"))
async def vertex_predict(request: Request):
    body = await request.json()
    return {"predictions": predict(body)}

import json
import os

import joblib
import pandas as pd

model = None


def init():                                    # Azure Machine Learning calls this once when the deployment starts
    global model
    model_dir = os.environ.get("AZUREML_MODEL_DIR", "model")
    model = joblib.load(os.path.join(model_dir, "model.joblib"))


def run(raw_data):                             # and this for every request to /score
    rows = json.loads(raw_data)["data"]
    frame = pd.DataFrame(rows, columns=list(model.feature_names_in_))
    return {"predictions": [round(float(v), 2) for v in model.predict(frame)]}

import os
import shutil
import subprocess
import sys
import time

import joblib
import pandas as pd
import requests
from sklearn.ensemble import HistGradientBoostingRegressor

mix = pd.read_csv("concrete.csv")                               # UCI Concrete Compressive Strength (CC BY 4.0)
features = [c for c in mix.columns if c != "strength_mpa"]
os.makedirs("model", exist_ok=True)
joblib.dump(HistGradientBoostingRegressor(random_state=0).fit(mix[features], mix.strength_mpa), "model/model.joblib")
rows = mix[features].head(3).values.tolist()

servers = [
    subprocess.Popen([sys.executable, "-m", "uvicorn", "app:app", "--port", "8080", "--log-level", "warning"],
                     env=dict(os.environ, MODEL_DIR="model", AIP_HEALTH_ROUTE="/health", AIP_PREDICT_ROUTE="/predict")),
    subprocess.Popen([shutil.which("azmlinfsrv"), "--entry_script", "score.py", "--port", "5005"],   # Azure's own server
                     env=dict(os.environ, AZUREML_MODEL_DIR="model"), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL),
]


def wait_for(url):
    for _ in range(120):
        try:
            if requests.get(url, timeout=1).ok:
                return
        except requests.ConnectionError:
            pass
        time.sleep(0.5)
    raise RuntimeError(f"{url} did not start")


try:
    for url in ("http://127.0.0.1:8080/ping", "http://127.0.0.1:8080/health", "http://127.0.0.1:5005/"):
        wait_for(url)
    calls = {"SageMaker AI  POST /invocations": ("http://127.0.0.1:8080/invocations", {"instances": rows}),
             "Google Cloud  POST /predict": ("http://127.0.0.1:8080/predict", {"instances": rows}),
             "Azure ML      POST /score": ("http://127.0.0.1:5005/score", {"data": rows})}
    answers = {}
    for name, (url, body) in calls.items():
        reply = requests.post(url, json=body, timeout=30)
        answers[name] = reply.json()["predictions"]
        print(f"{name:<34} {reply.status_code}  {answers[name]}")
    print(f"actual strengths (MPa): {mix.strength_mpa.head(3).round(2).tolist()}")
    print(f"same predictions from all three contracts: {len({tuple(a) for a in answers.values()}) == 1}")
    print("\nrequests in the wrong shape:")
    for name, url, body in (("SageMaker AI, 'data' instead of 'instances'", "http://127.0.0.1:8080/invocations", {"data": rows}),
                            ("Azure ML, 'instances' instead of 'data'", "http://127.0.0.1:5005/score", {"instances": rows})):
        print(f"  {name:<46} HTTP {requests.post(url, json=body, timeout=30).status_code}")
finally:
    for server in servers:
        if os.name == "nt":
            subprocess.run(["taskkill", "/T", "/F", "/PID", str(server.pid)], capture_output=True)
        else:
            server.terminate()

SageMaker AI  POST /invocations    200  [71.13, 68.29, 41.1]
Google Cloud  POST /predict        200  [71.13, 68.29, 41.1]
Azure ML      POST /score          200  [71.13, 68.29, 41.1]
actual strengths (MPa): [79.99, 61.89, 40.27]
same predictions from all three contracts: True

requests in the wrong shape:
  SageMaker AI, 'data' instead of 'instances'    HTTP 400
  Azure ML, 'instances' instead of 'data'        HTTP 500