Tutorial Lengkap AWS Lambda + SageMaker: Serverless ML Inference

Menggabungkan AWS Lambda dengan SageMaker memungkinkan inference ML yang serverless, scalable, dan hemat biaya. Tutorial ini mencakup pattern untuk mengintegrasikan fungsi Lambda dengan endpoint SageMaker untuk aplikasi ML production.

Mengapa Lambda + SageMaker?

• Serverless: Tidak perlu mengelola infrastruktur
• Hemat biaya: Bayar per invocation
• Scalable: Scaling otomatis sesuai demand
• Fleksibel: Multiple pattern integrasi
• Event-driven: Trigger inference dari event AWS apapun

• Real-time API inference
• Trigger batch processing
• Event-driven ML pipelines
• Inference hemat biaya
• Multi-model serving

Prerequisites

pip install boto3 sagemaker

AWS CLI sudah dikonfigurasi
aws configure

Pattern 1: Lambda Memanggil 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):
    """Panggil SageMaker endpoint untuk inference."""
    try:
        # Parse input
        if "body" in event:
            body = json.loads(event["body"])
        else:
            body = event

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

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

        # Panggil 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 dengan 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 dan kode
COPY model/ ${LAMBDATASKROOT}/model/

COPY lambdafunction.py ${LAMBDATASKROOT}/

CMD ["lambdafunction.lambdahandler"]

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

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

def lambdahandler(event, context):
    """Jalankan inference langsung di 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 dan push ke 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

Buat fungsi Lambda
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):
    """Proses pesan SQS untuk async inference."""
    results = []

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

        # Dapatkan input data dari S3
        inputbucket = message["inputbucket"]
        inputkey = message["inputkey"]

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

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

        # Panggil endpoint
        inferenceresponse = sagemakerruntime.invokeendpoint(

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

        )

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

        # Simpan hasil ke 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 ke model spesifik di 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 untuk Memulai 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')}"


    # Dapatkan parameter dari event
    inputpath = event.get("inputpath", "s3://bucket/input/")

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

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


    # Buat 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 fitur sebelum inference."""
    body = json.loads(event.get("body", "{}"))
    rawfeatures = body.get("features", [])

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

    scaledfeatures = SCALER.transform(features)

    # Panggil endpoint dengan fitur yang sudah dipreprocess
    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

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

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

)

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

def lambdahandler(event, context):

    """Inference robust dengan error handling."""
    try:
        body = json.loads(event.get("body", "{}"))
        features = body.get("features")

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

        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 dan Logging

1. CloudWatch Metrics

import boto3
from datetime import datetime

cloudwatch = boto3.client("cloudwatch")

def loginferencemetrics(endpointname, latency, success):

    """Log metrik custom ke 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 di luar handler
import boto3
sagemakerruntime = boto3.client("sagemaker-runtime")

def lambdahandler(event, context):

    # Gunakan client yang sudah diinisialisasi
    response = sagemakerruntime.invokeendpoint(...)

from botocore.config import Config

config = Config(
    maxpoolconnections=50,

    retries={"maxattempts": 3}
)

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

Kesimpulan

Lambda + SageMaker memungkinkan:

Key takeaways:

• Pilih pattern berdasarkan use case
• Optimasi untuk cold starts
• Implementasikan error handling yang proper
• Monitor latensi dan error
• Gunakan async untuk batch workloads