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 Workbench 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 Workbench Inherited Classes must implement the abstract transform_impl() method

Transform

Bases: ABC

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

Source code in src/workbench/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
142
143
144
145
146
147
148
149
150
151
class Transform(ABC):
    """Transform: Abstract Base Class for all transforms within Workbench. Inherited Classes
    must implement the abstract transform_impl() method"""

    def __init__(self, input_name: str, output_name: str, catalog_db: str = "workbench"):
        """Transform Initialization

        Args:
            input_name (str): The Name of the Input Artifact
            output_name (str): The Name of the Output Artifact
            catalog_db (str): The AWS Data Catalog Database to use (default: "workbench")
        """

        self.log = logging.getLogger("workbench")
        self.input_type = None
        self.output_type = None
        self.output_tags = ""
        self.input_name = str(input_name)  # Occasionally we get a pathlib.Path object
        self.output_name = str(output_name)  # Occasionally we get a pathlib.Path object
        self.output_meta = {"workbench_input": self.input_name}
        self.data_catalog_db = catalog_db

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

        # Grab a Workbench Role ARN, Boto3, SageMaker Session, and SageMaker Client
        self.aws_account_clamp = AWSAccountClamp()
        self.workbench_role_arn = self.aws_account_clamp.aws_session.get_workbench_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 V3 Tag format (list of Tag objects)"""
        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
        workbench_meta = {"workbench_tags": self.output_tags}
        for key, value in self.output_meta.items():
            workbench_meta[key] = value
        aws_tags = self.convert_to_aws_tags(workbench_meta)
        return aws_tags

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

        The output artifact is created as a side effect and registered under output_name; this method
        returns None. Retrieve the created artifact via its name (e.g. Model(output_name)), or use the
        higher-level API (e.g. FeatureSet.to_model()), which returns the artifact object directly.
        """
        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_name(self, input_name: str):
        """Set the Input Name (Name) for this Transform"""
        self.input_name = input_name

    def set_output_name(self, output_name: str):
        """Set the Output Name (Name) for this Transform"""
        self.output_name = output_name

__init__(input_name, output_name, catalog_db='workbench')

Transform Initialization

Parameters:

Name Type Description Default
input_name str

The Name of the Input Artifact

 required
output_name str

The Name of the Output Artifact

 required
catalog_db str

The AWS Data Catalog Database to use (default: "workbench")

 'workbench'
Source code in src/workbench/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
75
76
77
78
79
80
def __init__(self, input_name: str, output_name: str, catalog_db: str = "workbench"):
    """Transform Initialization

    Args:
        input_name (str): The Name of the Input Artifact
        output_name (str): The Name of the Output Artifact
        catalog_db (str): The AWS Data Catalog Database to use (default: "workbench")
    """

    self.log = logging.getLogger("workbench")
    self.input_type = None
    self.output_type = None
    self.output_tags = ""
    self.input_name = str(input_name)  # Occasionally we get a pathlib.Path object
    self.output_name = str(output_name)  # Occasionally we get a pathlib.Path object
    self.output_meta = {"workbench_input": self.input_name}
    self.data_catalog_db = catalog_db

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

    # Grab a Workbench Role ARN, Boto3, SageMaker Session, and SageMaker Client
    self.aws_account_clamp = AWSAccountClamp()
    self.workbench_role_arn = self.aws_account_clamp.aws_session.get_workbench_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/workbench/core/transforms/transform.py 
105
106
107
108
109
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 V3 Tag format (list of Tag objects)

Source code in src/workbench/core/transforms/transform.py 
111
112
113
114
@staticmethod
def convert_to_aws_tags(metadata: dict):
    """Convert a dictionary to the AWS V3 Tag format (list of Tag objects)"""
    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/workbench/core/transforms/transform.py 
116
117
118
119
120
121
122
123
def get_aws_tags(self):
    """Get the metadata/tags and convert them into AWS Tag Format"""
    # Set up our metadata storage
    workbench_meta = {"workbench_tags": self.output_tags}
    for key, value in self.output_meta.items():
        workbench_meta[key] = value
    aws_tags = self.convert_to_aws_tags(workbench_meta)
    return aws_tags

input_type()

What Input Type does this Transform Consume

Source code in src/workbench/core/transforms/transform.py 
137
138
139
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/workbench/core/transforms/transform.py 
141
142
143
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/workbench/core/transforms/transform.py 
91
92
93
94
@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/workbench/core/transforms/transform.py 
87
88
89
def pre_transform(self, **kwargs):
    """Perform any Pre-Transform operations"""
    self.log.debug("Pre-Transform...")

set_input_name(input_name)

Set the Input Name (Name) for this Transform

Source code in src/workbench/core/transforms/transform.py 
145
146
147
def set_input_name(self, input_name: str):
    """Set the Input Name (Name) for this Transform"""
    self.input_name = input_name

set_output_name(output_name)

Set the Output Name (Name) for this Transform

Source code in src/workbench/core/transforms/transform.py 
149
150
151
def set_output_name(self, output_name: str):
    """Set the Output Name (Name) for this Transform"""
    self.output_name = output_name

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/workbench/core/transforms/transform.py 
 96
 97
 98
 99
100
101
102
103
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

transform(**kwargs)

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

The output artifact is created as a side effect and registered under output_name; this method returns None. Retrieve the created artifact via its name (e.g. Model(output_name)), or use the higher-level API (e.g. FeatureSet.to_model()), which returns the artifact object directly.

Source code in src/workbench/core/transforms/transform.py 
125
126
127
128
129
130
131
132
133
134
135
@final
def transform(self, **kwargs):
    """Perform the Transformation from Input to Output with pre_transform() and post_transform() invocations.

    The output artifact is created as a side effect and registered under output_name; this method
    returns None. Retrieve the created artifact via its name (e.g. Model(output_name)), or use the
    higher-level API (e.g. FeatureSet.to_model()), which returns the artifact object directly.
    """
    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/workbench/core/transforms/transform.py 
82
83
84
85
@abstractmethod
def transform_impl(self, **kwargs):
    """Abstract Method: Implement the Transformation from Input to Output"""
    pass

TransformInput

Bases: Enum

Enumerated Types for Workbench Transform Inputs

Source code in src/workbench/core/transforms/transform.py 
14
15
16
17
18
19
20
21
22
23
class TransformInput(Enum):
    """Enumerated Types for Workbench 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 Workbench Transform Outputs

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

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