Skip to content

Transform

API Classes

The API Classes will use Transforms internally. So model.to_endpoint() uses the ModelToEndpoint() transform. If you need more control over the Transform you can use the Core Classes directly.

The SageWorks Transform class is a base/abstract class that defines API implemented by all the child classes (DataLoaders, DataSourceToFeatureSet, ModelToEndpoint, etc).

Transform: Base Class for all transforms within SageWorks Inherited Classes must implement the abstract transform_impl() method

Transform

Bases: ABC

Transform: Abstract Base Class for all transforms within SageWorks. Inherited Classes must implement the abstract transform_impl() method

Source code in src/sageworks/core/transforms/transform.py 
 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class Transform(ABC):
    """Transform: Abstract Base Class for all transforms within SageWorks. Inherited Classes
    must implement the abstract transform_impl() method"""

    def __init__(self, input_uuid: str, output_uuid: str):
        """Transform Initialization"""

        self.log = logging.getLogger("sageworks")
        self.input_type = None
        self.output_type = None
        self.output_tags = ""
        self.input_uuid = str(input_uuid)  # Occasionally we get a pathlib.Path object
        self.output_uuid = str(output_uuid)  # Occasionally we get a pathlib.Path object
        self.output_meta = {"sageworks_input": self.input_uuid}
        self.data_catalog_db = "sageworks"

        # Grab our SageWorks Bucket
        cm = ConfigManager()
        if not cm.config_okay():
            self.log.error("SageWorks Configuration Incomplete...")
            self.log.error("Run the 'sageworks' command and follow the prompts...")
            raise FatalConfigError()
        self.sageworks_bucket = cm.get_config("SAGEWORKS_BUCKET")
        self.data_sources_s3_path = "s3://" + self.sageworks_bucket + "/data-sources"
        self.feature_sets_s3_path = "s3://" + self.sageworks_bucket + "/feature-sets"
        self.models_s3_path = "s3://" + self.sageworks_bucket + "/models"
        self.endpoints_sets_s3_path = "s3://" + self.sageworks_bucket + "/endpoints"

        # Grab a SageWorks Role ARN, Boto3, SageMaker Session, and SageMaker Client
        self.aws_account_clamp = AWSAccountClamp()
        self.sageworks_role_arn = self.aws_account_clamp.aws_session.get_sageworks_execution_role_arn()
        self.boto3_session = self.aws_account_clamp.boto3_session
        self.sm_session = self.aws_account_clamp.sagemaker_session()
        self.sm_client = self.aws_account_clamp.sagemaker_client()

        # Delimiter for storing lists in AWS Tags
        self.tag_delimiter = "::"

    @abstractmethod
    def transform_impl(self, **kwargs):
        """Abstract Method: Implement the Transformation from Input to Output"""
        pass

    def pre_transform(self, **kwargs):
        """Perform any Pre-Transform operations"""
        self.log.debug("Pre-Transform...")

    @abstractmethod
    def post_transform(self, **kwargs):
        """Post-Transform ensures that the output Artifact is ready for use"""
        pass

    def set_output_tags(self, tags: Union[list, str]):
        """Set the tags that will be associated with the output object
        Args:
            tags (Union[list, str]): The list of tags or a '::' separated string of tags"""
        if isinstance(tags, list):
            self.output_tags = self.tag_delimiter.join(tags)
        else:
            self.output_tags = tags

    def add_output_meta(self, meta: dict):
        """Add additional metadata that will be associated with the output artifact
        Args:
            meta (dict): A dictionary of metadata"""
        self.output_meta = self.output_meta | meta

    @staticmethod
    def convert_to_aws_tags(metadata: dict):
        """Convert a dictionary to the AWS tag format (list of dicts)
        [ {Key: key_name, Value: value}, {..}, ...]"""
        return [{"Key": key, "Value": value} for key, value in metadata.items()]

    def get_aws_tags(self):
        """Get the metadata/tags and convert them into AWS Tag Format"""
        # Set up our metadata storage
        sageworks_meta = {"sageworks_tags": self.output_tags}
        for key, value in self.output_meta.items():
            sageworks_meta[key] = value
        aws_tags = self.convert_to_aws_tags(sageworks_meta)
        return aws_tags

    @final
    def transform(self, **kwargs):
        """Perform the Transformation from Input to Output with pre_transform() and post_transform() invocations"""
        self.pre_transform(**kwargs)
        self.transform_impl(**kwargs)
        self.post_transform(**kwargs)

    def input_type(self) -> TransformInput:
        """What Input Type does this Transform Consume"""
        return self.input_type

    def output_type(self) -> TransformOutput:
        """What Output Type does this Transform Produce"""
        return self.output_type

    def set_input_uuid(self, input_uuid: str):
        """Set the Input UUID (Name) for this Transform"""
        self.input_uuid = input_uuid

    def set_output_uuid(self, output_uuid: str):
        """Set the Output UUID (Name) for this Transform"""
        self.output_uuid = output_uuid

__init__(input_uuid, output_uuid)

Transform Initialization

Source code in src/sageworks/core/transforms/transform.py 
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
def __init__(self, input_uuid: str, output_uuid: str):
    """Transform Initialization"""

    self.log = logging.getLogger("sageworks")
    self.input_type = None
    self.output_type = None
    self.output_tags = ""
    self.input_uuid = str(input_uuid)  # Occasionally we get a pathlib.Path object
    self.output_uuid = str(output_uuid)  # Occasionally we get a pathlib.Path object
    self.output_meta = {"sageworks_input": self.input_uuid}
    self.data_catalog_db = "sageworks"

    # Grab our SageWorks Bucket
    cm = ConfigManager()
    if not cm.config_okay():
        self.log.error("SageWorks Configuration Incomplete...")
        self.log.error("Run the 'sageworks' command and follow the prompts...")
        raise FatalConfigError()
    self.sageworks_bucket = cm.get_config("SAGEWORKS_BUCKET")
    self.data_sources_s3_path = "s3://" + self.sageworks_bucket + "/data-sources"
    self.feature_sets_s3_path = "s3://" + self.sageworks_bucket + "/feature-sets"
    self.models_s3_path = "s3://" + self.sageworks_bucket + "/models"
    self.endpoints_sets_s3_path = "s3://" + self.sageworks_bucket + "/endpoints"

    # Grab a SageWorks Role ARN, Boto3, SageMaker Session, and SageMaker Client
    self.aws_account_clamp = AWSAccountClamp()
    self.sageworks_role_arn = self.aws_account_clamp.aws_session.get_sageworks_execution_role_arn()
    self.boto3_session = self.aws_account_clamp.boto3_session
    self.sm_session = self.aws_account_clamp.sagemaker_session()
    self.sm_client = self.aws_account_clamp.sagemaker_client()

    # Delimiter for storing lists in AWS Tags
    self.tag_delimiter = "::"

add_output_meta(meta)

Add additional metadata that will be associated with the output artifact Args: meta (dict): A dictionary of metadata

Source code in src/sageworks/core/transforms/transform.py 
 99
100
101
102
103
def add_output_meta(self, meta: dict):
    """Add additional metadata that will be associated with the output artifact
    Args:
        meta (dict): A dictionary of metadata"""
    self.output_meta = self.output_meta | meta

convert_to_aws_tags(metadata) staticmethod

Convert a dictionary to the AWS tag format (list of dicts) [ {Key: key_name, Value: value}, {..}, ...]

Source code in src/sageworks/core/transforms/transform.py 
105
106
107
108
109
@staticmethod
def convert_to_aws_tags(metadata: dict):
    """Convert a dictionary to the AWS tag format (list of dicts)
    [ {Key: key_name, Value: value}, {..}, ...]"""
    return [{"Key": key, "Value": value} for key, value in metadata.items()]

get_aws_tags()

Get the metadata/tags and convert them into AWS Tag Format

Source code in src/sageworks/core/transforms/transform.py 
111
112
113
114
115
116
117
118
def get_aws_tags(self):
    """Get the metadata/tags and convert them into AWS Tag Format"""
    # Set up our metadata storage
    sageworks_meta = {"sageworks_tags": self.output_tags}
    for key, value in self.output_meta.items():
        sageworks_meta[key] = value
    aws_tags = self.convert_to_aws_tags(sageworks_meta)
    return aws_tags

input_type()

What Input Type does this Transform Consume

Source code in src/sageworks/core/transforms/transform.py 
127
128
129
def input_type(self) -> TransformInput:
    """What Input Type does this Transform Consume"""
    return self.input_type

output_type()

What Output Type does this Transform Produce

Source code in src/sageworks/core/transforms/transform.py 
131
132
133
def output_type(self) -> TransformOutput:
    """What Output Type does this Transform Produce"""
    return self.output_type

post_transform(**kwargs) abstractmethod

Post-Transform ensures that the output Artifact is ready for use

Source code in src/sageworks/core/transforms/transform.py 
85
86
87
88
@abstractmethod
def post_transform(self, **kwargs):
    """Post-Transform ensures that the output Artifact is ready for use"""
    pass

pre_transform(**kwargs)

Perform any Pre-Transform operations

Source code in src/sageworks/core/transforms/transform.py 
81
82
83
def pre_transform(self, **kwargs):
    """Perform any Pre-Transform operations"""
    self.log.debug("Pre-Transform...")

set_input_uuid(input_uuid)

Set the Input UUID (Name) for this Transform

Source code in src/sageworks/core/transforms/transform.py 
135
136
137
def set_input_uuid(self, input_uuid: str):
    """Set the Input UUID (Name) for this Transform"""
    self.input_uuid = input_uuid

set_output_tags(tags)

Set the tags that will be associated with the output object Args: tags (Union[list, str]): The list of tags or a '::' separated string of tags

Source code in src/sageworks/core/transforms/transform.py 
90
91
92
93
94
95
96
97
def set_output_tags(self, tags: Union[list, str]):
    """Set the tags that will be associated with the output object
    Args:
        tags (Union[list, str]): The list of tags or a '::' separated string of tags"""
    if isinstance(tags, list):
        self.output_tags = self.tag_delimiter.join(tags)
    else:
        self.output_tags = tags

set_output_uuid(output_uuid)

Set the Output UUID (Name) for this Transform

Source code in src/sageworks/core/transforms/transform.py 
139
140
141
def set_output_uuid(self, output_uuid: str):
    """Set the Output UUID (Name) for this Transform"""
    self.output_uuid = output_uuid

transform(**kwargs)

Perform the Transformation from Input to Output with pre_transform() and post_transform() invocations

Source code in src/sageworks/core/transforms/transform.py 
120
121
122
123
124
125
@final
def transform(self, **kwargs):
    """Perform the Transformation from Input to Output with pre_transform() and post_transform() invocations"""
    self.pre_transform(**kwargs)
    self.transform_impl(**kwargs)
    self.post_transform(**kwargs)

transform_impl(**kwargs) abstractmethod

Abstract Method: Implement the Transformation from Input to Output

Source code in src/sageworks/core/transforms/transform.py 
76
77
78
79
@abstractmethod
def transform_impl(self, **kwargs):
    """Abstract Method: Implement the Transformation from Input to Output"""
    pass

TransformInput

Bases: Enum

Enumerated Types for SageWorks Transform Inputs

Source code in src/sageworks/core/transforms/transform.py 
14
15
16
17
18
19
20
21
22
23
class TransformInput(Enum):
    """Enumerated Types for SageWorks Transform Inputs"""

    LOCAL_FILE = auto()
    PANDAS_DF = auto()
    SPARK_DF = auto()
    S3_OBJECT = auto()
    DATA_SOURCE = auto()
    FEATURE_SET = auto()
    MODEL = auto()

TransformOutput

Bases: Enum

Enumerated Types for SageWorks Transform Outputs

Source code in src/sageworks/core/transforms/transform.py 
26
27
28
29
30
31
32
33
34
35
class TransformOutput(Enum):
    """Enumerated Types for SageWorks Transform Outputs"""

    PANDAS_DF = auto()
    SPARK_DF = auto()
    S3_OBJECT = auto()
    DATA_SOURCE = auto()
    FEATURE_SET = auto()
    MODEL = auto()
    ENDPOINT = auto()