flask模型部署的Sagemaker超时

Question

下面是ECR容器中的predict.py。Sagemaker端点在重试10-12分钟后输出"Status:Failed“。/ping和/invocations方法都可用

/opt/ml/code/predict.py
----------
logger = logging.getLogger()
logger.setLevel(logging.INFO)
classpath =  <.pkl file> 
model = pickle.load(open(classpath, "rb"))


app = flask.Flask(__name__)
print(app)

@app.route("/ping", methods=["GET"]
def ping():
    """Determine if the container is working and healthy."""
    return flask.Response(response="Flask running", status=200, mimetype="application/json")

@app.route("/invocations", methods=["POST"])
    ""InferenceCode""
    return flask.Response(response="Invocation Completed", status=200, 
    mimetype="application/json")

Below snippet was both added and removed , however I still have the endpoint in failed status

 if __name__ == '__main__':
     app.run(host='0.0.0.0',port=5000)

Error : 
"The primary container for production variant <modelname> did not pass the ping health check. Please check CloudWatch logs for this endpoint."


Sagemaker endpoint Cloudwatch logs.
[INFO] Starting gunicorn 20.1.0
[INFO] Listening at: http://0.0.0.0:8000 (1)
[INFO] Using worker: sync
[INFO] Booting worker with pid: 11```

Answer 1

您的predictor文件用于测试模型是否加载到/ping中，以及您是否可以在/invocations中执行推理。如果您已经在SageMaker上训练了模型，则需要从/opt/ml目录加载它，如下所示。

prefix = "/opt/ml/"
model_path = os.path.join(prefix, "model")

class ScoringService(object):
    model = None  # Where we keep the model when it's loaded

    @classmethod
    def get_model(rgrs):
        """Get the model object for this instance, loading it if it's not already loaded."""
        if rgrs.model == None:
            with open(os.path.join(model_path, "rf-model.pkl"), "rb") as inp:
                rgrs.model = pickle.load(inp)
        return rgrs.model

    @classmethod
    def predict(rgrs, input):
        """For the input, do the predictions and return them.
        Args:
            input (a pandas dataframe): The data on which to do the predictions. There will be
                one prediction per row in the dataframe"""
        rf = rgrs.get_model()
        return rf.predict(input)

该类帮助加载您的模型，然后我们可以在/ping中进行验证。

# The flask app for serving predictions
app = flask.Flask(__name__)


@app.route("/ping", methods=["GET"])
def ping():
    """Determine if the container is working and healthy. In this sample container, we declare
    it healthy if we can load the model successfully."""
    health = ScoringService.get_model() is not None  # You can insert a health check here

    status = 200 if health else 404
    return flask.Response(response="\n", status=status, mimetype="application/json")

在这里，SageMaker将测试您是否正确加载了模型。对于/invocations，包含要传递到模型的预测功能中的任何数据格式的推理逻辑。

@app.route("/invocations", methods=["POST"])
def transformation():
    
    data = None

    # Convert from CSV to pandas
    if flask.request.content_type == "text/csv":
        data = flask.request.data.decode("utf-8")
        s = io.StringIO(data)
        data = pd.read_csv(s, header=None)
    else:
        return flask.Response(
            response="This predictor only supports CSV data", status=415, mimetype="text/plain"
        )

    print("Invoked with {} records".format(data.shape[0]))

    # Do the prediction
    predictions = ScoringService.predict(data)

    # Convert from numpy back to CSV
    out = io.StringIO()
    pd.DataFrame({"results": predictions}).to_csv(out, header=False, index=False)
    result = out.getvalue()
    
    
    return flask.Response(response=result, status=200, mimetype="text/csv")

确保如上所示设置或配置您的predictor.py，以便SageMaker能够正确地理解/检索您的模型。

我在AWS工作&我的观点是我自己的。