Skip to content

Model to Endpoint

API Classes

For most users the API Classes will provide all the general functionality to create a full AWS ML Pipeline

ModelToEndpoint: Deploy an Endpoint for a Model

ModelToEndpoint

Bases: Transform

ModelToEndpoint: Deploy an Endpoint for a Model

Common Usage 
to_endpoint = ModelToEndpoint(model_name, endpoint_name)
to_endpoint.set_output_tags(["aqsol", "public", "whatever"])
to_endpoint.transform()
Source code in src/workbench/core/transforms/model_to_endpoint/model_to_endpoint.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class ModelToEndpoint(Transform):
    """ModelToEndpoint: Deploy an Endpoint for a Model

    Common Usage:
        ```python
        to_endpoint = ModelToEndpoint(model_name, endpoint_name)
        to_endpoint.set_output_tags(["aqsol", "public", "whatever"])
        to_endpoint.transform()
        ```
    """

    def __init__(self, model_name: str, endpoint_name: str, serverless: bool = True, instance: str = "ml.t2.medium"):
        """ModelToEndpoint Initialization
        Args:
            model_name(str): The Name of the input Model
            endpoint_name(str): The Name of the output Endpoint
            serverless(bool): Deploy the Endpoint in serverless mode (default: True)
            instance(str): The instance type to use for the Endpoint (default: "ml.t2.medium")
        """
        # Make sure the endpoint_name is a valid name
        Artifact.is_name_valid(endpoint_name, delimiter="-", lower_case=False)

        # Call superclass init
        super().__init__(model_name, endpoint_name)

        # Set up all my instance attributes
        self.serverless = serverless
        self.instance_type = "serverless" if serverless else instance
        self.input_type = TransformInput.MODEL
        self.output_type = TransformOutput.ENDPOINT

    def transform_impl(self, **kwargs):
        """Deploy an Endpoint for a Model"""

        # Delete endpoint (if it already exists)
        EndpointCore.managed_delete(self.output_name)

        # Get the Model Package ARN for our input model
        input_model = ModelCore(self.input_name)
        model_package_arn = input_model.model_package_arn()

        # Deploy the model
        self._deploy_model(model_package_arn, **kwargs)

        # Add this endpoint to the set of registered endpoints for the model
        input_model.register_endpoint(self.output_name)

        # This ensures that the endpoint is ready for use
        time.sleep(5)  # We wait for AWS Lag
        end = EndpointCore(self.output_name)
        self.log.important(f"Endpoint {end.name} is ready for use")

    def _deploy_model(self, model_package_arn: str, mem_size: int = 2048, max_concurrency: int = 5):
        """Internal Method: Deploy the Model

        Args:
            model_package_arn(str): The Model Package ARN used to deploy the Endpoint
        """
        # Grab the specified Model Package
        model_package = ModelPackage(
            role=self.workbench_role_arn,
            model_package_arn=model_package_arn,
            sagemaker_session=self.sm_session,
        )

        # Log the image that will be used for deployment
        inference_image = self.sm_client.describe_model_package(ModelPackageName=model_package_arn)[
            "InferenceSpecification"
        ]["Containers"][0]["Image"]
        self.log.important(f"Deploying Model Package: {self.input_name} with Inference Image: {inference_image}")

        # Get the metadata/tags to push into AWS
        aws_tags = self.get_aws_tags()

        # Is this a serverless deployment?
        serverless_config = None
        if self.serverless:
            serverless_config = ServerlessInferenceConfig(
                memory_size_in_mb=mem_size,
                max_concurrency=max_concurrency,
            )

        # Deploy the Endpoint
        self.log.important(f"Deploying the Endpoint {self.output_name}...")
        model_package.deploy(
            initial_instance_count=1,
            instance_type=self.instance_type,
            serverless_inference_config=serverless_config,
            endpoint_name=self.output_name,
            serializer=CSVSerializer(),
            deserializer=CSVDeserializer(),
            tags=aws_tags,
        )

    def post_transform(self, **kwargs):
        """Post-Transform: Calling onboard() for the Endpoint"""
        self.log.info("Post-Transform: Calling onboard() for the Endpoint...")

        # Onboard the Endpoint
        output_endpoint = EndpointCore(self.output_name)
        output_endpoint.onboard_with_args(input_model=self.input_name)

__init__(model_name, endpoint_name, serverless=True, instance='ml.t2.medium')

ModelToEndpoint Initialization Args: model_name(str): The Name of the input Model endpoint_name(str): The Name of the output Endpoint serverless(bool): Deploy the Endpoint in serverless mode (default: True) instance(str): The instance type to use for the Endpoint (default: "ml.t2.medium")

Source code in src/workbench/core/transforms/model_to_endpoint/model_to_endpoint.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(self, model_name: str, endpoint_name: str, serverless: bool = True, instance: str = "ml.t2.medium"):
    """ModelToEndpoint Initialization
    Args:
        model_name(str): The Name of the input Model
        endpoint_name(str): The Name of the output Endpoint
        serverless(bool): Deploy the Endpoint in serverless mode (default: True)
        instance(str): The instance type to use for the Endpoint (default: "ml.t2.medium")
    """
    # Make sure the endpoint_name is a valid name
    Artifact.is_name_valid(endpoint_name, delimiter="-", lower_case=False)

    # Call superclass init
    super().__init__(model_name, endpoint_name)

    # Set up all my instance attributes
    self.serverless = serverless
    self.instance_type = "serverless" if serverless else instance
    self.input_type = TransformInput.MODEL
    self.output_type = TransformOutput.ENDPOINT

post_transform(**kwargs)

Post-Transform: Calling onboard() for the Endpoint

Source code in src/workbench/core/transforms/model_to_endpoint/model_to_endpoint.py 
110
111
112
113
114
115
116
def post_transform(self, **kwargs):
    """Post-Transform: Calling onboard() for the Endpoint"""
    self.log.info("Post-Transform: Calling onboard() for the Endpoint...")

    # Onboard the Endpoint
    output_endpoint = EndpointCore(self.output_name)
    output_endpoint.onboard_with_args(input_model=self.input_name)

transform_impl(**kwargs)

Deploy an Endpoint for a Model

Source code in src/workbench/core/transforms/model_to_endpoint/model_to_endpoint.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def transform_impl(self, **kwargs):
    """Deploy an Endpoint for a Model"""

    # Delete endpoint (if it already exists)
    EndpointCore.managed_delete(self.output_name)

    # Get the Model Package ARN for our input model
    input_model = ModelCore(self.input_name)
    model_package_arn = input_model.model_package_arn()

    # Deploy the model
    self._deploy_model(model_package_arn, **kwargs)

    # Add this endpoint to the set of registered endpoints for the model
    input_model.register_endpoint(self.output_name)

    # This ensures that the endpoint is ready for use
    time.sleep(5)  # We wait for AWS Lag
    end = EndpointCore(self.output_name)
    self.log.important(f"Endpoint {end.name} is ready for use")