-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathround.py
More file actions
160 lines (137 loc) · 5.35 KB
/
Copy pathround.py
File metadata and controls
160 lines (137 loc) · 5.35 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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
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
152
153
154
155
156
157
158
159
160
# Copyright [2024] Expedia, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# pylint: disable=unused-argument
# pylint: disable=invalid-name
# pylint: disable=too-many-ancestors
# pylint: disable=no-member
from typing import List, Optional
import keras
import pyspark.sql.functions as F
from pyspark import keyword_only
from pyspark.ml.param import Param, Params, TypeConverters
from pyspark.sql import DataFrame
from pyspark.sql.types import DataType, DoubleType, FloatType
from kamae.keras.core.backend import ALL_BACKENDS
from kamae.keras.core.layers import RoundLayer
from kamae.spark.params import SingleInputSingleOutputParams
from kamae.spark.utils import single_input_single_output_scalar_transform
from .base import BaseTransformer
class RoundParams(Params):
"""
Mixin class containing roundType parameter needed for rounding transform layers.
"""
roundType = Param(
Params._dummy(),
"roundType",
"Round type to use in round transform, one of 'floor', 'ceil' or 'round'.",
typeConverter=TypeConverters.toString,
)
def setRoundType(self, value: str) -> "RoundParams":
"""
Sets the roundType parameter.
:param value: Rounding type to use in round transform,
one of 'floor', 'ceil' or 'round'.
:returns: Instance of class mixed in.
"""
if value not in ["floor", "ceil", "round"]:
raise ValueError("roundType must be one of 'floor', 'ceil' or 'round'")
return self._set(roundType=value)
def getRoundType(self) -> str:
"""
Gets the roundType parameter.
:returns: Rounding type to use in round transform,
one of 'floor', 'ceil' or 'round'.
"""
return self.getOrDefault(self.roundType)
class RoundTransformer(
BaseTransformer,
SingleInputSingleOutputParams,
RoundParams,
):
"""
Round Spark Transformer for use in Spark pipelines.
This transformer rounds the input column to the nearest integer using the
specified rounding type.
"""
supported_backends = ALL_BACKENDS
jit_compatible = True
@keyword_only
def __init__(
self,
inputCol: Optional[str] = None,
outputCol: Optional[str] = None,
inputDtype: Optional[str] = None,
outputDtype: Optional[str] = None,
layerName: Optional[str] = None,
roundType: str = "round",
) -> None:
"""
Initializes an RoundTransformer transformer.
:param inputCol: Input column name.
:param outputCol: Output column name.
:param inputDtype: Input data type to cast input column to before
transforming.
:param outputDtype: Output data type to cast the output column to after
transforming.
:param layerName: Name of the layer. Used as the name of the Keras layer
in the keras model. If not set, we use the uid of the Spark transformer.
:param roundType: Rounding type to use in round transform,
one of 'floor', 'ceil' or 'round'. Defaults to 'round'.
:returns: None - class instantiated.
"""
super().__init__()
kwargs = self._input_kwargs
self._setDefault(roundType="round")
self.setParams(**kwargs)
@property
def compatible_dtypes(self) -> Optional[List[DataType]]:
"""
List of compatible data types for the layer.
If the computation can be performed on any data type, return None.
:returns: List of compatible data types for the layer.
"""
return [FloatType(), DoubleType()]
def _transform(self, dataset: DataFrame) -> DataFrame:
"""
Transforms the input dataset. Creates a new column with name `outputCol`,
which applies the rounding operation to the input column.
:param dataset: Pyspark dataframe to transform.
:returns: Transformed pyspark dataframe.
"""
func_dict = {
"floor": F.floor,
"ceil": F.ceil,
"round": F.round,
}
input_datatype = self.get_column_datatype(
dataset=dataset, column_name=self.getInputCol()
)
output_col = single_input_single_output_scalar_transform(
input_col=F.col(self.getInputCol()),
input_col_datatype=input_datatype,
func=lambda x: func_dict[self.getRoundType()](x),
)
return dataset.withColumn(self.getOutputCol(), output_col)
def get_keras_layer(self) -> keras.layers.Layer:
"""
Gets the Keras layer for the round transformer.
:returns: Keras layer with name equal to the layerName parameter that
performs a rounding operation.
"""
return RoundLayer(
name=self.getLayerName(),
input_dtype=self.getInputKerasDtype(),
output_dtype=self.getOutputKerasDtype(),
round_type=self.getRoundType(),
)