Complete AWS Lambda + SageMaker Tutorial: Serverless ML Inference

Combining AWS Lambda with SageMaker enables serverless, scalable, and cost-effective ML inference. This tutorial covers patterns for integrating Lambda functions with SageMaker endpoints for production ML applications.

Why Lambda + SageMaker?

• Serverless: No infrastructure management
• Cost-effective: Pay per invocation
• Scalable: Automatic scaling to demand
• Flexible: Multiple integration patterns
• Event-driven: Trigger inference from any AWS event

• Real-time API inference
• Batch processing triggers
• Event-driven ML pipelines
• Cost-optimized inference
• Multi-model serving

Prerequisites

pip install boto3 sagemaker

AWS CLI configured
aws configure

Pattern 1: Lambda Invoking SageMaker Endpoint

1. Deploy SageMaker Endpoint

import sagemaker
from sagemaker.sklearn import SKLearnModel

session = sagemaker.Session()
role = sagemaker.getexecutionrole()

Deploy model
model = SKLearnModel(
    modeldata="s3://bucket/model/model.tar.gz",

    role=role,
    frameworkversion="1.0-1",
    pyversion="py3"

)

predictor = model.deploy(
    initialinstancecount=1,

    instancetype="ml.m5.large",
    endpointname="sklearn-endpoint"

)

# lambdafunction.py
import json
import boto3
import os

Initialize SageMaker runtime client
sagemakerruntime = boto3.client("sagemaker-runtime")


ENDPOINTNAME = os.environ.get("SAGEMAKERENDPOINT", "sklearn-endpoint")


def lambdahandler(event, context):
    """Invoke SageMaker endpoint for inference."""
    try:
        # Parse input
        if "body" in event:
            body = json.loads(event["body"])
        else:
            body = event

        features = body.get("features", [])

        # Prepare payload
        payload = json.dumps({"features": features})

        # Invoke endpoint
        response = sagemakerruntime.invokeendpoint(
            EndpointName=ENDPOINTNAME,

            ContentType="application/json",
            Body=payload
        )

        # Parse response
        result = json.loads(response["Body"].read().decode())

        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({
                "prediction": result,
                "endpoint": ENDPOINTNAME
            })
        }

    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

3. Deploy Lambda with SAM

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  InferenceFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambdafunction.lambdahandler
      Runtime: python3.9
      Timeout: 30
      MemorySize: 256
      Environment:
        Variables:
          SAGEMAKERENDPOINT: sklearn-endpoint

      Policies:
Version: '2012-10-17'
          Statement:

Effect: Allow
              Action:

sagemaker:InvokeEndpoint
              Resource: '*'

      Events:
        Api:
          Type: Api
          Properties:
            Path: /predict
            Method: post

# Deploy
sam build
sam deploy --guided

# Dockerfile
FROM public.ecr.aws/lambda/python:3.9

Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

Copy model and code
COPY model/ ${LAMBDATASKROOT}/model/

COPY lambdafunction.py ${LAMBDATASKROOT}/

CMD ["lambdafunction.lambdahandler"]

# lambdafunction.py
import json
import joblib
import numpy as np

Load model at cold start
MODEL = joblib.load("model/model.joblib")

def lambdahandler(event, context):
    """Run inference directly in Lambda."""
    try:
        body = json.loads(event.get("body", "{}"))
        features = np.array(body["features"]).reshape(1, -1)

        prediction = MODEL.predict(features)
        probability = MODEL.predictproba(features)


        return {
            "statusCode": 200,
            "body": json.dumps({
                "prediction": int(prediction[0]),
                "probability": probability[0].tolist()
            })
        }

    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

# Build container
docker build -t ml-lambda .

Tag and push to ECR
aws ecr create-repository --repository-name ml-lambda
docker tag ml-lambda:latest 123456789.dkr.ecr.us-east-1.amazonaws.com/ml-lambda:latest
docker push 123456789.dkr.ecr.us-east-1.amazonaws.com/ml-lambda:latest

Create Lambda function
aws lambda create-function \
    --function-name ml-inference \
    --package-type Image \
    --code ImageUri=123456789.dkr.ecr.us-east-1.amazonaws.com/ml-lambda:latest \
    --role arn:aws:iam::123456789:role/lambda-role \
    --memory-size 1024 \
    --timeout 30

# asyncinference.py
import json
import boto3

sagemakerruntime = boto3.client("sagemaker-runtime")

s3 = boto3.client("s3")

def lambdahandler(event, context):
    """Process SQS messages for async inference."""
    results = []

    for record in event["Records"]:
        message = json.loads(record["body"])

        # Get input data from S3
        inputbucket = message["inputbucket"]
        inputkey = message["inputkey"]

        response = s3.getobject(Bucket=inputbucket, Key=inputkey)

        inputdata = response["Body"].read().decode()

        # Invoke endpoint
        inferenceresponse = sagemakerruntime.invokeendpoint(

            EndpointName=message["endpointname"],
            ContentType="text/csv",
            Body=inputdata

        )

        result = inferenceresponse["Body"].read().decode()

        # Save result to S3
        outputkey = message["outputkey"]
        s3.putobject(

            Bucket=inputbucket,
            Key=outputkey,

            Body=result
        )

        results.append({
            "input": inputkey,
            "output": outputkey,

            "status": "success"
        })

    return {"processed": len(results), "results": results}

# template.yaml
Resources:
  InferenceQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: inference-queue
      VisibilityTimeout: 300

  AsyncInferenceFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: asyncinference.lambdahandler

      Runtime: python3.9
      Timeout: 300
      MemorySize: 512
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt InferenceQueue.Arn
            BatchSize: 10

# multimodelinference.py
import json
import boto3

sagemakerruntime = boto3.client("sagemaker-runtime")

ENDPOINTNAME = "multi-model-endpoint"


def lambdahandler(event, context):
    """Route to specific model in multi-model endpoint."""
    body = json.loads(event.get("body", "{}"))

    modelname = body.get("model", "default-model.tar.gz")

    features = body.get("features", [])

    response = sagemakerruntime.invokeendpoint(

        EndpointName=ENDPOINTNAME,
        TargetModel=modelname,

        ContentType="application/json",
        Body=json.dumps({"features": features})
    )

    result = json.loads(response["Body"].read().decode())

    return {
        "statusCode": 200,
        "body": json.dumps({
            "model": modelname,
            "prediction": result
        })
    }

Pattern 5: Batch Transform Trigger

1. Lambda to Start Batch Transform

# batchtransform.py
import json
import boto3
from datetime import datetime

sagemaker = boto3.client("sagemaker")

def lambdahandler(event, context):
    """Trigger SageMaker batch transform job."""
    jobname = f"batch-transform-{datetime.now().strftime('%Y%m%d-%H%M%S')}"


    # Get parameters from event
    inputpath = event.get("inputpath", "s3://bucket/input/")

    outputpath = event.get("outputpath", "s3://bucket/output/")

    modelname = event.get("modelname", "my-model")


    # Create transform job
    response = sagemaker.createtransformjob(

        TransformJobName=jobname,
        ModelName=modelname,

        TransformInput={
            "DataSource": {
                "S3DataSource": {
                    "S3DataType": "S3Prefix",
                    "S3Uri": inputpath
                }
            },
            "ContentType": "text/csv",
            "SplitType": "Line"
        },
        TransformOutput={
            "S3OutputPath": outputpath

        },
        TransformResources={
            "InstanceType": "ml.m5.large",
            "InstanceCount": 1
        }
    )

    return {
        "statusCode": 200,
        "body": json.dumps({
            "jobname": jobname,

            "status": "Started"
        })
    }

Resources:
  S3TriggerRule:
    Type: AWS::Events::Rule
    Properties:
      EventPattern:
        source:
aws.s3
        detail-type:

Object Created
        detail:

          bucket:
            name:
my-input-bucket
          object:

            key:
prefix: "batch-input/"
      Targets:

Id: BatchTransformLambda
          Arn: !GetAtt BatchTransformFunction.Arn

# preprocessing.py
import json
import boto3
import numpy as np
from sklearn.preprocessing import StandardScaler
import joblib

Load scaler
SCALER = joblib.load("scaler.joblib")
sagemakerruntime = boto3.client("sagemaker-runtime")

def lambdahandler(event, context):

    """Preprocess features before inference."""
    body = json.loads(event.get("body", "{}"))
    rawfeatures = body.get("features", [])

    # Preprocess
    features = np.array(rawfeatures).reshape(1, -1)

    scaledfeatures = SCALER.transform(features)

    # Invoke endpoint with preprocessed features
    response = sagemakerruntime.invokeendpoint(
        EndpointName="my-endpoint",
        ContentType="application/json",
        Body=json.dumps({"features": scaledfeatures.tolist()})

    )

    result = json.loads(response["Body"].read().decode())

    return {
        "statusCode": 200,
        "body": json.dumps({
            "rawfeatures": rawfeatures,

            "prediction": result
        })
    }

# robustinference.py
import json
import boto3
from botocore.config import Config

Configure retries
config = Config(
    retries={
        "maxattempts": 3,

        "mode": "adaptive"
    },
    connecttimeout=5,
    readtimeout=30

)

sagemakerruntime = boto3.client("sagemaker-runtime", config=config)

def lambdahandler(event, context):

    """Robust inference with error handling."""
    try:
        body = json.loads(event.get("body", "{}"))
        features = body.get("features")

        if not features:
            return {
                "statusCode": 400,
                "body": json.dumps({"error": "Missing features"})
            }

        response = sagemakerruntime.invokeendpoint(

            EndpointName="my-endpoint",
            ContentType="application/json",
            Body=json.dumps({"features": features})
        )

        result = json.loads(response["Body"].read().decode())

        return {
            "statusCode": 200,
            "body": json.dumps({"prediction": result})
        }

    except sagemakerruntime.exceptions.ModelError as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "Model error", "details": str(e)})
        }

    except Exception as e:
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

Monitoring and Logging

1. CloudWatch Metrics

import boto3
from datetime import datetime

cloudwatch = boto3.client("cloudwatch")

def loginferencemetrics(endpointname, latency, success):

    """Log custom metrics to CloudWatch."""
    cloudwatch.putmetricdata(

        Namespace="ML/Inference",
        MetricData=[
            {
                "MetricName": "InferenceLatency",
                "Dimensions": [
                    {"Name": "Endpoint", "Value": endpointname}
                ],
                "Value": latency,
                "Unit": "Milliseconds"
            },
            {
                "MetricName": "InferenceSuccess",
                "Dimensions": [
                    {"Name": "Endpoint", "Value": endpointname}

                ],
                "Value": 1 if success else 0,
                "Unit": "Count"
            }
        ]
    )

# Initialize outside handler
import boto3
sagemakerruntime = boto3.client("sagemaker-runtime")

def lambdahandler(event, context):

    # Use pre-initialized client
    response = sagemakerruntime.invokeendpoint(...)

from botocore.config import Config

config = Config(
    maxpoolconnections=50,

    retries={"maxattempts": 3}
)

client = boto3.client("sagemaker-runtime", config=config)

Conclusion

Lambda + SageMaker enables:

Key takeaways:

• Choose pattern based on use case
• Optimize for cold starts
• Implement proper error handling
• Monitor latency and errors
• Use async for batch workloads