AWS，或称亚马逊网络服务（Amazon Web Services），是许多企业用于存储、分析、应用、部署服务等领域的云计算服务。这是一个利用多个服务以按使用量付费方式来支持企业的无服务器方式的平台。
机器学习建模活动也是AWS支持的活动之一。利用多种服务，可以支持建模活动，如开发模型到将其投入生产。AWS展现了在需要可扩展性和速度的任何企业中至关重要的多样性。
本文将讨论在AWS云环境中部署机器学习模型并投入生产。我们该如何做到这一点？让我们进一步探索。
准备工作
在开始本教程之前，您需要创建一个AWS账户，因为我们需要使用它们来访问所有AWS服务。我假设读者将使用免费套餐来参考本文。此外，我假设读者已经了解如何使用Python编程语言，并具备基本的机器学习知识。此外，我们将重点关注模型部署部分，不会集中精力研究数据预处理和模型评估等数据科学活动的其他方面。
有了这个想法，我们就开始了将机器学习模型部署到AWS云服务的旅程。
在AWS上部署模型
在本教程中，我们将开发一个机器学习模型来预测流失情况。训练数据集是从Kaggle获取的，您可以在此处下载。
在获取了数据集之后，我们将创建一个S3存储桶来存储数据集。在AWS服务中搜索S3并创建存储桶。
在本文中，我将存储桶命名为“telecom-churn-dataset”并将其设置为新加坡位置。如果您愿意，您可以更改这些设置，但暂时让我们使用这个设置。
创建完存储桶并将数据上传到存储桶后，我们将进入AWS SageMaker服务。在这项服务中，我们将使用Studio作为我们的工作环境。如果您以前从未使用过Studio，请在继续之前创建一个域和用户。
首先，在Amazon SageMaker管理配置中选择域。
在域中，您会看到许多按钮可供选择。在此屏幕上，选择创建域按钮。
如果想加快创建过程，请选择快速设置。完成后，您应该在仪表板中看到一个新创建的域。选择刚刚创建的新域，然后点击添加用户按钮。
接下来，您应该根据自己的喜好命名用户配置文件。至于执行角色，您现在可以保持默认设置，因为这是在域创建过程中创建的角色。
一路点击下一步，直到画布设置。在此部分，我关闭了我们不需要的几个设置，如时间序列预测。
完成所有设置后，进入Studio选择并选择带有刚刚创建的用户名的打开Studio按钮。
在Studio中，导航到类似文件夹图标的侧边栏，并在那里创建一个新的notebook。我们可以使用默认设置，如下图所示。
作者：WIJAYA
借助新的笔记本电脑，我们将创建一个流失预测模型，并将该模型部署到API推理中，以便在生产环境中使用。
首先，让我们导入必要的包并读取流失数据。
import boto3
import pandas as pd
import sagemaker
sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()
df = pd.read_csv('s3://telecom-churn-dataset/telecom_churn.csv') 
作者：WIJAYA
接下来，我们将使用以下代码将上述数据分割为训练数据和测试数据。
from sklearn.model_selection import train_test_split
train, test = train_test_split(df, test_size=0.3, random_state=42)
我们将测试数据设定为原始数据的30%。有了数据分割后，我们将把它们重新上传到S3存储桶中。
bucket = 'telecom-churn-dataset'
train.to_csv(f's3://{bucket}/telecom_churn_train.csv', index=False)
test.to_csv(f's3://{bucket}/telecom_churn_test.csv', index=False)
您可以在您的S3存储桶中查看数据，目前包含三个不同的数据集。
作者：WIJAYA
准备好我们的数据集后，我们将开发一个流失预测模型并将其部署。在AWS中，我们经常使用脚本训练方法进行机器学习训练。所以在开始训练之前，我们需要先开发一个脚本。
下一步，我们需要在同一个文件夹中创建一个额外的Python文件，我称之为train.py。
作者：WIJAYA
在此文件中，我们将设置模型开发过程以创建流失模型。对于本教程，我将采用Ram Vegiraju的一些代码。
首先，我们将导入开发模型所需的所有必要包。
import argparse
import os
import io
import boto3
import json
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import joblib
接下来，我们将使用解析器方法来控制我们在训练过程中可以输入的变量。我们要放入脚本中以训练我们的模型的整体代码如下。
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--estimator', type=int, default=10)
parser.add_argument('--sm-model-dir', type=str, default=os.environ.get('SM_MODEL_DIR'))
parser.add_argument('--model_dir', type=str)
parser.add_argument('--train', type=str, default=os.environ.get('SM_CHANNEL_TRAIN'))
args, _ = parser.parse_known_args()
estimator = args.estimator
model_dir = args.model_dir
sm_model_dir = args.sm_model_dir
training_dir = args.train
s3_client = boto3.client('s3')
bucket = 'telecom-churn-dataset'
obj = s3_client.get_object(Bucket=bucket, Key='telecom_churn_train.csv')
train_data = pd.read_csv(io.BytesIO(obj['Body'].read()))
obj = s3_client.get_object(Bucket=bucket, Key='telecom_churn_test.csv')
test_data = pd.read_csv(io.BytesIO(obj['Body'].read()))
X_train = train_data.drop('Churn', axis=1)
X_test = test_data.drop('Churn', axis=1)
y_train = train_data['Churn']
y_test = test_data['Churn']
rfc = RandomForestClassifier(n_estimators=estimator)
rfc.fit(X_train, y_train)
y_pred = rfc.predict(X_test)
print('准确率：',accuracy_score(y_test, y_pred))
joblib.dump(rfc, os.path.join(args.sm_model_dir, "rfc_model.joblib"))
最后，我们需要放置SageMaker要求进行推断所需的四个不同的函数：model_fn，input_fn，output_fn和predict_fn。
#反序列化模型以加载它们def model_fn(model_dir): model = joblib.load(os.path.join(model_dir, "rfc_model.joblib")) return model#应用程序的请求输入def input_fn(request_body, request_content_type): if request_content_type == 'application/json': request_body = json.loads(request_body) inp_var = request_body['Input'] return inp_var else: raise ValueError("此模型仅支持application/json输入")#预测函数def predict_fn(input_data, model): return model.predict(input_data)#输出函数def output_fn(prediction, content_type): res = int(prediction[0]) resJSON = {'Output': res} return resJSON
有了准备好的脚本，我们将运行训练过程。在下一步中，我们将把上面创建的脚本传递给SKLearn估计器。这个估计器是一个Sagemaker对象，它将处理整个训练过程，我们只需要传递与以下代码类似的所有参数。
from sagemaker.sklearn import SKLearnsklearn_estimator = SKLearn(entry_point='train.py', role=role, instance_count=1, instance_type='ml.c4.2xlarge', py_version='py3', framework_version='0.23-1', script_mode=True, hyperparameters={ 'estimator': 15})sklearn_estimator.fit()
如果训练成功，您将得到以下报告。
Image by Author
如果您想要检查SKLearn训练的Docker镜像以及您的模型工件位置，您可以使用以下代码访问它们。
model_artifact = sklearn_estimator.model_dataimage_uri = sklearn_estimator.image_uriprint(f'模型工件保存在：{model_artifact}')print(f'图像URI为：{image_uri}')
将模型放置到位置之后，我们将部署该模型到一个API端点，以便用于预测。为此，我们可以使用以下代码。
import timechurn_endpoint_name='churn-rf-model-'+time.strftime("%Y-%m-%d-%H-%M-%S", time.gmtime())churn_predictor=sklearn_estimator.deploy(initial_instance_count=1,instance_type='ml.m5.large',endpoint_name=churn_endpoint_name)
如果部署成功，将创建模型端点，您可以访问它以创建预测。您还可以在Sagemaker仪表板中看到该端点。
Image by Author
现在可以使用此端点进行预测。为此，您可以使用以下代码测试端点。
import boto3.client('sagemaker-runtime')content_type = "application/json"#用您打算使用的输入数据替换request_body = {"Input": [[128,1,1,2.70,1,265.1,110,89.0, 9.87,10.0]]}#用您的端点名称替换endpoint_name = "churn-rf-model-2023-09-24-12-29-04" #数据序列化data = json.loads(json.dumps(request_body))payload = json.dumps(data)#调用端点response = client.invoke_endpoint( EndpointName=endpoint_name, ContentType=content_type, Body=payload)result = json.loads(response['Body'].read().decode())['Output']result
恭喜。您现在已成功在AWS Cloud中部署了模型。完成测试过程后，请不要忘记清理端点。您可以使用以下代码执行此操作。
from sagemaker import Sessionsagemaker_session = Session()sagemaker_session.delete_endpoint(endpoint_name='您的端点名称')
不要忘记关闭您使用的实例，并清理不再需要的S3存储。
如果您希望不使用端点模型，您可以阅读更多关于SKLearn estimator和Batch Transform inferences的文章。
结论
AWS云平台是许多公司用来支持业务的多功能平台。其中一个经常使用的服务是用于数据分析目的，尤其是模型生产。在本文中，我们学习了如何使用AWS SageMaker以及如何将模型部署到端点。Cornellius Yudha Wijaya是一名数据科学助理经理和数据作者。他在Allianz Indonesia全职工作期间，喜欢通过社交媒体和写作媒体分享Python和数据技巧。