<jupyter_start><jupyter_text>Stress TestThe idea of this code is to see how the production Endpoint will behave when a **bunch** of requests arrive it.Let's simulate several users doing predictions at the same time<jupyter_code>import threading
import boto3
import numpy as np
import time
import math

from multiprocessing.pool import ThreadPool
from sklearn import datasets
sm = boto3.client("sagemaker-runtime")

endpoint_name_mask='iris-model-%s'

iris = datasets.load_iris()
dataset = np.insert(iris.data, 0, iris.target,axis=1)
from sagemaker.serializers import CSVSerializer

def predict(payload):
    csv_serializer = CSVSerializer()
    payload = payload
    X = payload[1:]
    y = payload[0]
    
    elapsed_time = time.time()
    resp = sm.invoke_endpoint(
        EndpointName=endpoint_name_mask % env,
        ContentType='text/csv',
        Accept='text/csv',
        Body=csv_serializer.serialize(X)
    )
    elapsed_time = time.time() - elapsed_time
    resp = float(resp['Body'].read().decode('utf-8').strip())
    return (resp == y, elapsed_time)
def run_test(max_threads, max_requests):
    num_batches = math.ceil(max_requests / len(dataset))
    requests = []
    for i in range(num_batches):
        batch = dataset.copy()
        np.random.shuffle(batch)
        requests += batch.tolist()
    len(requests)

    pool = ThreadPool(max_threads)
    result = pool.map(predict, requests)
    pool.close()
    pool.join()
    
    correct_random_forest=0
    elapsedtime_random_forest=0
    for i in result:
        correct_random_forest += i[0]
        elapsedtime_random_forest += i[1]
    print("Score classifier: {}".format(correct_random_forest/len(result)))

    print("Elapsed time: {}s".format(elapsedtime_random_forest))
env='production'
%%time
print("Starting test 1")
run_test(10, 1000)
%%time
print("Starting test 2")
run_test(100, 10000)
%%time
print("Starting test 3")
run_test(150, 100000000)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup