#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2019 The FATE Authors. All Rights Reserved.
#
# 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.
#
################################################################################
#
#
################################################################################
import copy
import functools
import numpy as np
from federatedml.feature.instance import Instance
from federatedml.feature.sparse_vector import SparseVector
from federatedml.model_base import ModelBase
from federatedml.protobuf.generated.data_transform_meta_pb2 import DataTransformMeta
from federatedml.protobuf.generated.data_transform_meta_pb2 import DataTransformImputerMeta
from federatedml.protobuf.generated.data_transform_meta_pb2 import DataTransformOutlierMeta
from federatedml.protobuf.generated.data_transform_param_pb2 import DataTransformParam
from federatedml.protobuf.generated.data_transform_param_pb2 import DataTransformImputerParam
from federatedml.protobuf.generated.data_transform_param_pb2 import DataTransformOutlierParam
from federatedml.statistic import data_overview
from federatedml.util import abnormal_detection
from federatedml.util import consts
from federatedml.util import LOGGER
from federatedml.util.io_check import assert_io_num_rows_equal
from federatedml.util.data_format_preprocess import DataFormatPreProcess
from federatedml.util.anonymous_generator_util import Anonymous
# =============================================================================
# DenseFeatureTransformer
# =============================================================================
class DenseFeatureTransformer(object):
def __init__(self, data_transform_param):
self.delimitor = data_transform_param.delimitor
self.data_type = data_transform_param.data_type
self.missing_fill = data_transform_param.missing_fill
self.default_value = data_transform_param.default_value
self.missing_fill_method = data_transform_param.missing_fill_method
self.missing_impute = data_transform_param.missing_impute
self.outlier_replace = data_transform_param.outlier_replace
self.outlier_replace_method = data_transform_param.outlier_replace_method
self.outlier_impute = data_transform_param.outlier_impute
self.outlier_replace_value = data_transform_param.outlier_replace_value
self.with_label = data_transform_param.with_label
self.label_name = data_transform_param.label_name.lower() if self.with_label else None
self.label_type = data_transform_param.label_type if self.with_label else None
self.output_format = data_transform_param.output_format
self.missing_impute_rate = None
self.outlier_replace_rate = None
self.header = None
self.sid_name = None
self.exclusive_data_type_fid_map = {}
self.match_id_name = data_transform_param.match_id_name
self.match_id_index = 0
self.with_match_id = data_transform_param.with_match_id
self.anonymous_generator = None
self.anonymous_header = None
if data_transform_param.exclusive_data_type:
self.exclusive_data_type = dict([(k.lower(), v)
for k, v in data_transform_param.exclusive_data_type.items()])
else:
self.exclusive_data_type = None
def _update_param(self, schema):
meta = schema["meta"]
self.delimitor = meta.get("delimiter", ",")
self.data_type = meta.get("data_type")
self.with_label = meta.get("with_label", False)
if self.with_label:
self.label_type = meta.get("label_type", "int")
self.label_name = meta.get("label_name", '')
self.with_match_id = meta.get("with_match_id", False)
if self.with_match_id:
match_id_name = schema.get("match_id_name", [])
if not self.match_id_name:
if isinstance(match_id_name, list) and len(self.match_id_name) > 1:
raise ValueError("Multiple Match ID exist, please specified the one to use")
self.match_id_name = match_id_name[0] if isinstance(match_id_name, list) else match_id_name
self.match_id_index = schema["original_index_info"]["match_id_index"][0]
else:
try:
idx = match_id_name.index(self.match_id_name)
except ValueError:
raise ValueError(f"Can not find {self.match_id_name} in {match_id_name}")
self.match_id_index = schema["original_index_info"]["match_id_index"][idx]
schema["match_id_name"] = self.match_id_name
header = schema["header"]
exclusive_data_type = meta.get("exclusive_data_type", None)
if exclusive_data_type:
self.exclusive_data_type = dict([(k.lower(), v) for k, v in exclusive_data_type.items()])
for idx, col_name in enumerate(header):
if col_name in self.exclusive_data_type:
self.exclusive_data_type_fid_map[idx] = self.exclusive_data_type[col_name]
def extract_feature_value(self, value, header_index=None):
if not header_index:
return []
value = value.split(self.delimitor, -1)
if len(value) <= header_index[-1]:
raise ValueError("Feature shape is smaller than header shape")
feature_values = []
for idx in header_index:
feature_values.append(value[idx])
return feature_values
def read_data(self, input_data, mode="fit"):
LOGGER.info("start to read dense data and change data to instance")
abnormal_detection.empty_table_detection(input_data)
schema = copy.deepcopy(input_data.schema)
if not schema.get("meta"):
LOGGER.warning("Data meta is supported to be set with data uploading or binding, "
"please refer to data transform using guides.")
meta = dict(input_format="dense",
delimiter=self.delimitor,
with_label=self.with_label,
label_name=self.label_name,
with_match_id=self.with_match_id,
data_type=self.data_type,
)
if mode == "transform" and self.with_label \
and self.label_name not in schema["header"].split(self.delimitor, -1):
del meta["label_name"]
del meta["with_label"]
schema["meta"] = meta
generated_header = DataFormatPreProcess.generate_header(input_data, schema)
schema.update(generated_header)
schema = self.anonymous_generator.generate_anonymous_header(schema)
set_schema(input_data, schema)
else:
self._update_param(schema)
header = schema["header"]
anonymous_header = schema["anonymous_header"]
training_header = self.header
if mode == "transform":
if (set(self.header) & set(header)) != set(self.header):
raise ValueError(f"Transform Data's header is {header}, expect {self.header}")
self.header = header
if not self.anonymous_header:
self.anonymous_header = anonymous_header
else:
self.header = header
self.anonymous_header = anonymous_header
header_index = schema["original_index_info"]["header_index"]
extract_feature_func = functools.partial(self.extract_feature_value,
header_index=header_index)
input_data_features = input_data.mapValues(extract_feature_func)
# input_data_features.schema = input_data.schema
input_data_features.schema = schema
input_data_labels = None
input_data_match_id = None
if "label_name" in schema:
label_index = schema["original_index_info"]["label_index"]
input_data_labels = input_data.mapValues(lambda value: value.split(self.delimitor, -1)[label_index])
if self.with_match_id:
input_data_match_id = input_data.mapValues(
lambda value: value.split(self.delimitor, -1)[self.match_id_index])
if mode == "fit":
data_instance = self.fit(input_data, input_data_features, input_data_labels, input_data_match_id)
set_schema(data_instance, schema)
else:
data_instance = self.transform(input_data_features, input_data_labels, input_data_match_id)
data_instance = data_overview.header_alignment(data_instance, training_header, self.anonymous_header)
self.header = training_header
return data_instance
def fit(self, input_data, input_data_features, input_data_labels, input_data_match_id):
input_data_features = self.fill_missing_value(input_data_features, "fit")
input_data_features = self.replace_outlier_value(input_data_features, "fit")
data_instance = self.gen_data_instance(input_data_features, input_data_labels, input_data_match_id)
return data_instance
@assert_io_num_rows_equal
def transform(self, input_data_features, input_data_labels, input_data_match_id):
schema = input_data_features.schema
input_data_features = self.fill_missing_value(input_data_features, "transform")
input_data_features = self.replace_outlier_value(input_data_features, "transform")
data_instance = self.gen_data_instance(input_data_features, input_data_labels, input_data_match_id)
data_instance.schema = schema
return data_instance
def fill_missing_value(self, input_data_features, mode="fit"):
if self.missing_fill:
from federatedml.feature.imputer import Imputer
imputer_processor = Imputer(self.missing_impute)
if mode == "fit":
input_data_features, self.default_value = imputer_processor.fit(input_data_features,
replace_method=self.missing_fill_method,
replace_value=self.default_value)
if self.missing_impute is None:
self.missing_impute = imputer_processor.get_missing_value_list()
else:
input_data_features = imputer_processor.transform(input_data_features,
transform_value=self.default_value)
if self.missing_impute is None:
self.missing_impute = imputer_processor.get_missing_value_list()
self.missing_impute_rate = imputer_processor.get_impute_rate(mode)
return input_data_features
def replace_outlier_value(self, input_data_features, mode="fit"):
if self.outlier_replace:
from federatedml.feature.imputer import Imputer
imputer_processor = Imputer(self.outlier_impute)
if mode == "fit":
input_data_features, self.outlier_replace_value = \
imputer_processor.fit(input_data_features,
replace_method=self.outlier_replace_method,
replace_value=self.outlier_replace_value)
if self.outlier_impute is None:
self.outlier_impute = imputer_processor.get_missing_value_list()
else:
input_data_features = imputer_processor.transform(input_data_features,
transform_value=self.outlier_replace_value)
self.outlier_replace_rate = imputer_processor.get_impute_rate(mode)
return input_data_features
def gen_data_instance(self, input_data_features, input_data_labels, input_data_match_id):
if input_data_labels:
data_instance = input_data_features.join(input_data_labels,
lambda features, label: self.to_instance(features, label))
else:
data_instance = input_data_features.mapValues(lambda features: self.to_instance(features))
if self.with_match_id:
data_instance = data_instance.join(input_data_match_id, self.append_match_id)
return data_instance
def append_match_id(self, inst, match_id):
inst.inst_id = match_id
return inst
def to_instance(self, features, label=None):
if self.header is None and len(features) != 0:
raise ValueError("features shape {} not equal to header shape 0".format(len(features)))
elif self.header is not None and len(self.header) != len(features):
raise ValueError("features shape {} not equal to header shape {}".format(len(features), len(self.header)))
if label is not None:
if self.label_type == 'int':
label = int(label)
elif self.label_type in ["float", "float64"]:
label = float(label)
format_features = DenseFeatureTransformer.gen_output_format(features, self.data_type,
self.exclusive_data_type_fid_map,
self.output_format,
missing_impute=self.missing_impute)
else:
format_features = DenseFeatureTransformer.gen_output_format(features, self.data_type,
self.exclusive_data_type_fid_map,
self.output_format,
missing_impute=self.missing_impute)
return Instance(inst_id=None,
features=format_features,
label=label)
@staticmethod
def gen_output_format(features, data_type='float', exclusive_data_type_fid_map=None,
output_format='dense', missing_impute=None):
if output_format not in ["dense", "sparse"]:
raise ValueError("output format {} is not define".format(output_format))
missing_impute_dtype_set = {"int", "int64", "long", "float", "float64", "double"}
missing_impute_value_set = {'', 'NULL', 'null', "NA"}
type_mapping = dict()
if output_format == "dense":
# format_features = copy.deepcopy(features)
format_features = [None] * len(features)
for fid in range(len(features)):
if exclusive_data_type_fid_map is not None and fid in exclusive_data_type_fid_map:
dtype = exclusive_data_type_fid_map[fid]
else:
dtype = data_type
if dtype in missing_impute_dtype_set:
if (missing_impute is not None and features[fid] in missing_impute) or \
(missing_impute is None and features[fid] in missing_impute_value_set):
format_features[fid] = np.nan
continue
format_features[fid] = features[fid]
if exclusive_data_type_fid_map:
if dtype not in type_mapping:
np_type = getattr(np, dtype)
type_mapping[dtype] = np_type
format_features[fid] = type_mapping[dtype](format_features[fid])
if exclusive_data_type_fid_map:
return np.asarray(format_features, dtype=object)
else:
return np.asarray(format_features, dtype=data_type)
indices = []
data = []
column_shape = len(features)
non_zero = 0
for i in range(column_shape):
if (missing_impute is not None and features[i] in missing_impute) or \
(missing_impute is None and features[i] in missing_impute_value_set):
indices.append(i)
data.append(np.nan)
non_zero += 1
elif data_type in ['float', 'float64', "double"]:
if np.fabs(float(features[i])) < consts.FLOAT_ZERO:
continue
indices.append(i)
data.append(float(features[i]))
non_zero += 1
elif data_type in ['int', "int64", "long"]:
if int(features[i]) == 0:
continue
indices.append(i)
data.append(int(features[i]))
else:
indices.append(i)
data.append(features[i])
return SparseVector(indices, data, column_shape)
def get_summary(self):
if not self.missing_fill and not self.outlier_replace:
return {}
summary_buf = {}
if self.missing_fill:
missing_summary = dict()
missing_summary["missing_value"] = list(self.missing_impute)
missing_summary["missing_impute_value"] = dict(zip(self.header, self.default_value))
missing_summary["missing_impute_rate"] = dict(zip(self.header, self.missing_impute_rate))
summary_buf["missing_fill_info"] = missing_summary
if self.outlier_replace:
outlier_replace_summary = dict()
outlier_replace_summary["outlier_value"] = list(self.outlier_impute)
outlier_replace_summary["outlier_replace_value"] = dict(zip(self.header, self.outlier_replace_value))
outlier_replace_summary["outlier_replace_rate"] = dict(zip(self.header, self.outlier_replace_rate))
summary_buf["outlier_replace_rate"] = outlier_replace_summary
return summary_buf
def save_model(self):
transform_meta, transform_param = save_data_transform_model(input_format="dense",
delimitor=self.delimitor,
data_type=self.data_type,
exclusive_data_type=self.exclusive_data_type,
with_label=self.with_label,
label_type=self.label_type,
output_format=self.output_format,
header=self.header,
sid_name=self.sid_name,
label_name=self.label_name,
with_match_id=self.with_match_id,
model_name="DenseFeatureTransformer",
anonymous_header=self.anonymous_header)
missing_imputer_meta, missing_imputer_param = save_missing_imputer_model(self.missing_fill,
self.missing_fill_method,
self.missing_impute,
self.default_value,
self.missing_impute_rate,
self.header,
"Imputer")
transform_meta.imputer_meta.CopyFrom(missing_imputer_meta)
transform_param.imputer_param.CopyFrom(missing_imputer_param)
outlier_meta, outlier_param = save_outlier_model(self.outlier_replace,
self.outlier_replace_method,
self.outlier_impute,
self.outlier_replace_value,
self.outlier_replace_rate,
self.header,
"Outlier")
transform_meta.outlier_meta.CopyFrom(outlier_meta)
transform_param.outlier_param.CopyFrom(outlier_param)
return {"DataTransformMeta": transform_meta,
"DataTransformParam": transform_param
}
def load_model(self, model_meta, model_param):
self.delimitor, self.data_type, self.exclusive_data_type, _1, _2, self.with_label, \
self.label_type, self.output_format, self.header, self.sid_name, self.label_name, self.with_match_id, self.anonymous_header = \
load_data_transform_model("DenseFeatureTransformer", model_meta, model_param)
self.missing_fill, self.missing_fill_method, \
self.missing_impute, self.default_value = load_missing_imputer_model(self.header,
"Imputer",
model_meta.imputer_meta,
model_param.imputer_param)
self.outlier_replace, self.outlier_replace_method, \
self.outlier_impute, self.outlier_replace_value = load_outlier_model(self.header,
"Outlier",
model_meta.outlier_meta,
model_param.outlier_param)
# =============================================================================
# SparseFeatureTransformer: mainly for libsvm input format
# =============================================================================
class SparseFeatureTransformer(object):
def __init__(self, data_transform_param):
self.delimitor = data_transform_param.delimitor
self.data_type = data_transform_param.data_type
self.label_type = data_transform_param.label_type
self.output_format = data_transform_param.output_format
self.header = None
self.sid_name = "sid"
self.with_match_id = data_transform_param.with_match_id
self.match_id_name = "match_id" if self.with_match_id else None
self.match_id_index = data_transform_param.match_id_index
self.with_label = data_transform_param.with_label
self.label_name = data_transform_param.label_name.lower() if self.with_label else None
self.anonymous_generator = None
self.anonymous_header = None
def _update_param(self, schema):
meta = schema["meta"]
self.delimitor = meta.get("delimiter", ",")
self.data_type = meta.get("data_type")
self.with_label = meta.get("with_label", False)
if self.with_label:
self.label_type = meta.get("label_type", "int")
self.label_name = meta.get("label_name", "")
self.with_match_id = meta.get("with_match_id", False)
if self.with_match_id:
match_id_name = schema.get("match_id_name")
if isinstance(match_id_name, list):
self.match_id_name = match_id_name[self.match_id_index]
else:
self.match_id_name = match_id_name
schema["match_id_name"] = self.match_id_name
def read_data(self, input_data, mode="fit"):
LOGGER.info("start to read sparse data and change data to instance")
abnormal_detection.empty_table_detection(input_data)
schema = copy.deepcopy(input_data.schema)
if not schema.get("meta", {}):
LOGGER.warning("Data meta is supported to be set with data uploading or binding, "
"please refer to data transform using guides.")
meta = dict(input_format="sparse",
delimiter=self.delimitor,
with_label=self.with_label,
with_match_id=self.with_match_id,
data_type=self.data_type)
schema["meta"] = meta
generated_header = DataFormatPreProcess.generate_header(input_data, schema)
schema.update(generated_header)
schema = self.anonymous_generator.generate_anonymous_header(schema)
set_schema(input_data, schema)
else:
self._update_param(schema)
if mode == "fit":
self.header = schema["header"]
self.anonymous_header = schema["anonymous_header"]
data_instance = self.fit(input_data)
else:
if not self.anonymous_header:
header_set = set(self.header)
self.anonymous_header = []
for column, anonymous_column in zip(schema["header"], schema["anonymous_header"]):
if column not in header_set:
continue
self.anonymous_header.append(anonymous_column)
schema["header"] = self.header
schema["anonymous_header"] = self.anonymous_header
set_schema(input_data, schema)
data_instance = self.transform(input_data)
set_schema(data_instance, schema)
return data_instance
def fit(self, input_data):
max_feature = len(self.header)
if max_feature == 0:
raise ValueError("no feature value in input data, please check!")
data_instance = self.gen_data_instance(input_data, max_feature)
return data_instance
def transform(self, input_data):
max_feature = len(self.header)
data_instance = self.gen_data_instance(input_data, max_feature)
return data_instance
def gen_data_instance(self, input_data, max_feature):
id_range = input_data.schema["meta"].get("id_range", 0)
params = [self.delimitor, self.data_type,
self.label_type, self.with_match_id,
self.match_id_index, id_range,
self.output_format,
self.with_label, max_feature]
to_instance_with_param = functools.partial(self.to_instance, params)
data_instance = input_data.mapValues(to_instance_with_param)
return data_instance
@staticmethod
def to_instance(param_list, value):
delimitor = param_list[0]
data_type = param_list[1]
label_type = param_list[2]
with_match_id = param_list[3]
match_id_index = param_list[4]
id_range = param_list[5]
output_format = param_list[6]
with_label = param_list[7]
max_fid = param_list[8]
if output_format not in ["dense", "sparse"]:
raise ValueError("output format {} is not define".format(output_format))
cols = value.split(delimitor, -1)
offset = 0
if with_match_id:
offset = id_range if id_range else 1
match_id = cols[match_id_index]
else:
match_id = None
label = None
if with_label:
label = cols[offset]
if label_type == 'int':
label = int(label)
elif label_type in ["float", "float64"]:
label = float(label)
offset += 1
fid_value = []
for i in range(offset, len(cols)):
fid, val = cols[i].split(":", -1)
fid = int(fid)
if data_type in ["float", "float64"]:
val = float(val)
elif data_type in ["int", "int64"]:
val = int(val)
fid_value.append((fid, val))
if output_format == "dense":
features = [0 for i in range(max_fid)]
for fid, val in fid_value:
features[fid] = val
features = np.asarray(features, dtype=data_type)
else:
indices = []
data = []
for fid, val in fid_value:
indices.append(fid)
data.append(val)
features = SparseVector(indices, data, max_fid)
return Instance(inst_id=match_id,
features=features,
label=label)
def save_model(self):
transform_meta, transform_param = save_data_transform_model(input_format="sparse",
delimitor=self.delimitor,
data_type=self.data_type,
label_type=self.label_type,
output_format=self.output_format,
header=self.header,
sid_name=self.sid_name,
label_name=self.label_name,
with_match_id=self.with_match_id,
with_label=self.with_label,
model_name="SparseFeatureTransformer",
anonymous_header=self.anonymous_header)
missing_imputer_meta, missing_imputer_param = save_missing_imputer_model(missing_fill=False,
model_name="Imputer")
transform_meta.imputer_meta.CopyFrom(missing_imputer_meta)
transform_param.imputer_param.CopyFrom(missing_imputer_param)
outlier_meta, outlier_param = save_outlier_model(outlier_replace=False,
model_name="Outlier")
transform_meta.outlier_meta.CopyFrom(outlier_meta)
transform_param.outlier_param.CopyFrom(outlier_param)
return {"DataTransformMeta": transform_meta,
"DataTransformParam": transform_param
}
def load_model(self, model_meta, model_param):
self.delimitor, self.data_type, _0, _1, _2, self.with_label, self.label_type, self.output_format, \
self.header, self.sid_name, self.label_name, self.with_match_id, self.anonymous_header = \
load_data_transform_model(
"SparseFeatureTransformer",
model_meta,
model_param)
# =============================================================================
# SparseTagTransformer: mainly for tag data
# =============================================================================
class SparseTagTransformer(object):
def __init__(self, data_transform_param):
self.delimitor = data_transform_param.delimitor
self.data_type = data_transform_param.data_type
self.tag_with_value = data_transform_param.tag_with_value
self.tag_value_delimitor = data_transform_param.tag_value_delimitor
self.with_label = data_transform_param.with_label
self.label_type = data_transform_param.label_type if self.with_label else None
self.output_format = data_transform_param.output_format
self.header = None
self.sid_name = "sid"
self.label_name = data_transform_param.label_name.lower() if data_transform_param.label_name else None
self.missing_fill = data_transform_param.missing_fill
self.missing_fill_method = data_transform_param.missing_fill_method
self.default_value = data_transform_param.default_value
self.with_match_id = data_transform_param.with_match_id
self.match_id_index = data_transform_param.match_id_index
self.match_id_name = "match_id" if self.with_match_id else None
self.missing_impute_rate = None
self.missing_impute = None
self.anonymous_generator = None
self.anonymous_header = None
def _update_param(self, schema):
meta = schema["meta"]
self.delimitor = meta.get("delimiter", ",")
self.data_type = meta.get("data_type")
self.tag_with_value = meta.get("tag_with_value")
self.tag_value_delimitor = meta.get("tag_value_delimiter", ":")
self.with_label = meta.get("with_label", False)
if self.with_label:
self.label_type = meta.get("label_type", "int")
self.label_name = meta.get("label_name")
self.with_match_id = meta.get("with_match_id", False)
if self.with_match_id:
match_id_name = schema.get("match_id_name")
if isinstance(match_id_name, list):
if not isinstance(self.match_id_index, int) or self.match_id_index >= len(match_id_name):
raise ValueError(f"match id index should between 0 and {len(match_id_name) - 1}, "
f"but {self.match_id_index} is given")
self.match_id_name = match_id_name[self.match_id_index]
else:
if self.match_id_index != 0:
raise ValueError("Only one match_id exist, match_id_index should be 0")
self.match_id_name = match_id_name
schema["match_id_name"] = self.match_id_name
def read_data(self, input_data, mode="fit"):
LOGGER.info("start to read sparse data and change data to instance")
abnormal_detection.empty_table_detection(input_data)
schema = copy.deepcopy(input_data.schema)
if not schema.get("meta", {}):
LOGGER.warning("Data meta is supported to be set with data uploading or binding, "
"please refer to data transform using guides.")
meta = dict(input_format="tag",
delimiter=self.delimitor,
with_label=self.with_label,
with_match_id=self.with_match_id,
tag_with_value=self.tag_with_value,
tag_value_delimiter=self.tag_value_delimitor,
data_type=self.data_type)
schema["meta"] = meta
generated_header = DataFormatPreProcess.generate_header(input_data, schema)
schema.update(generated_header)
schema = self.anonymous_generator.generate_anonymous_header(schema)
set_schema(input_data, schema)
else:
self._update_param(schema)
if mode == "fit":
self.header = schema["header"]
self.anonymous_header = schema["anonymous_header"]
data_instance = self.fit(input_data)
else:
if not self.anonymous_header:
header_set = set(self.header)
self.anonymous_header = []
for column, anonymous_column in zip(schema["header"], schema["anonymous_header"]):
if column not in header_set:
continue
self.anonymous_header.append(anonymous_column)
schema["header"] = self.header
schema["anonymous_header"] = self.anonymous_header
set_schema(input_data, schema)
data_instance = self.transform(input_data)
set_schema(data_instance, schema)
return data_instance
@staticmethod
def change_tag_to_str(value, tags_dict=None, delimitor=",", feature_offset=0,
tag_value_delimitor=":"):
vals = value.split(delimitor, -1)
ret = [''] * len(tags_dict)
vals = vals[feature_offset:]
for i in range(len(vals)):
tag, value = vals[i].split(tag_value_delimitor, -1)
idx = tags_dict.get(tag, None)
if idx is not None:
ret[idx] = value
return ret
@staticmethod
def change_str_to_tag(value, tags_dict=None, delimitor=",", tag_value_delimitor=":"):
ret = [None] * len(tags_dict)
tags = sorted(list(tags_dict.keys()))
for i in range(len(value)):
tag, val = tags[i], value[i]
ret[i] = tag_value_delimitor.join([tag, val])
return delimitor.join(ret)
def fill_missing_value(self, input_data, tags_dict, schema, mode="fit"):
feature_offset = DataFormatPreProcess.get_feature_offset(schema)
str_trans_method = functools.partial(self.change_tag_to_str,
tags_dict=tags_dict,
delimitor=self.delimitor,
feature_offset=feature_offset,
tag_value_delimitor=self.tag_value_delimitor)
input_data = input_data.mapValues(str_trans_method)
set_schema(input_data, schema)
from federatedml.feature.imputer import Imputer
imputer_processor = Imputer()
if mode == "fit":
data, self.default_value = imputer_processor.fit(input_data,
replace_method=self.missing_fill_method,
replace_value=self.default_value)
LOGGER.debug("self.default_value is {}".format(self.default_value))
else:
data = imputer_processor.transform(input_data,
transform_value=self.default_value)
if self.missing_impute is None:
self.missing_impute = imputer_processor.get_missing_value_list()
LOGGER.debug("self.missing_impute is {}".format(self.missing_impute))
self.missing_impute_rate = imputer_processor.get_impute_rate(mode)
str_trans_tag_method = functools.partial(self.change_str_to_tag,
tags_dict=tags_dict,
delimitor=self.delimitor,
tag_value_delimitor=self.tag_value_delimitor)
data = data.mapValues(str_trans_tag_method)
return data
def fit(self, input_data):
schema = input_data.schema
tags_dict = dict(zip(schema["header"], range(len(schema["header"]))))
if self.tag_with_value and self.missing_fill:
input_data = self.fill_missing_value(input_data, tags_dict, schema, mode="fit")
data_instance = self.gen_data_instance(input_data, schema["meta"], tags_dict)
return data_instance
def transform(self, input_data):
schema = input_data.schema
tags_dict = dict(zip(self.header, range(len(self.header))))
if self.tag_with_value and self.missing_fill:
input_data = self.fill_missing_value(input_data, tags_dict, schema, mode="transform")
data_instance = self.gen_data_instance(input_data, schema["meta"], tags_dict)
return data_instance
def gen_data_instance(self, input_data, meta, tags_dict):
params = [self.delimitor,
self.data_type,
self.tag_with_value,
self.tag_value_delimitor,
self.with_label,
self.with_match_id,
self.match_id_index,
meta.get("id_range", 0),
self.label_type,
self.output_format,
tags_dict]
to_instance_with_param = functools.partial(self.to_instance, params)
data_instance = input_data.mapValues(to_instance_with_param)
return data_instance
def get_summary(self):
if not self.missing_fill:
return {}
missing_summary = dict()
missing_summary["missing_value"] = list(self.missing_impute)
missing_summary["missing_impute_value"] = dict(zip(self.header, self.default_value))
missing_summary["missing_impute_rate"] = dict(zip(self.header, self.missing_impute_rate))
summary_buf = {"missing_fill_info": missing_summary}
return summary_buf
@staticmethod
def to_instance(param_list, value):
delimitor = param_list[0]
data_type = param_list[1]
tag_with_value = param_list[2]
tag_value_delimitor = param_list[3]
with_label = param_list[4]
with_match_id = param_list[5]
match_id_index = param_list[6]
id_range = param_list[7]
label_type = param_list[8]
output_format = param_list[9]
tags_dict = param_list[10]
if output_format not in ["dense", "sparse"]:
raise ValueError("output format {} is not define".format(output_format))
cols = value.split(delimitor, -1)
offset = 0
label = None
match_id = None
if with_match_id:
offset = id_range if id_range else 1
if offset == 0:
offset = 1
match_id = cols[match_id_index]
if with_label:
label = cols[offset]
offset += 1
if label_type == 'int':
label = int(label)
elif label_type in ["float", "float64"]:
label = float(label)
if output_format == "dense":
features = [0 for i in range(len(tags_dict))]
for fea in cols[offset:]:
if tag_with_value:
_tag, _val = fea.split(tag_value_delimitor, -1)
if _tag in tags_dict:
features[tags_dict.get(_tag)] = _val
else:
if fea in tags_dict:
features[tags_dict.get(fea)] = 1
features = np.asarray(features, dtype=data_type)
else:
indices = []
data = []
for fea in cols[offset:]:
if tag_with_value:
_tag, _val = fea.split(tag_value_delimitor, -1)
else:
_tag = fea
_val = 1
if _tag not in tags_dict:
continue
indices.append(tags_dict.get(_tag))
if data_type in ["float", "float64"]:
_val = float(_val)
elif data_type in ["int", "int64", "long"]:
_val = int(_val)
elif data_type == "str":
_val = str(_val)
data.append(_val)
features = SparseVector(indices, data, len(tags_dict))
return Instance(inst_id=match_id,
features=features,
label=label)
def save_model(self):
transform_meta, transform_param = save_data_transform_model(input_format="tag",
delimitor=self.delimitor,
data_type=self.data_type,
tag_with_value=self.tag_with_value,
tag_value_delimitor=self.tag_value_delimitor,
with_label=self.with_label,
label_type=self.label_type,
with_match_id=self.with_match_id,
output_format=self.output_format,
header=self.header,
sid_name=self.sid_name,
label_name=self.label_name,
model_name="Transformer",
anonymous_header=self.anonymous_header)
missing_imputer_meta, missing_imputer_param = save_missing_imputer_model(self.missing_fill,
self.missing_fill_method,
self.missing_impute,
self.default_value,
self.missing_impute_rate,
self.header,
"Imputer")
transform_meta.imputer_meta.CopyFrom(missing_imputer_meta)
transform_param.imputer_param.CopyFrom(missing_imputer_param)
outlier_meta, outlier_param = save_outlier_model(outlier_replace=False,
model_name="Outlier")
transform_meta.outlier_meta.CopyFrom(outlier_meta)
transform_param.outlier_param.CopyFrom(outlier_param)
return {"DataTransformMeta": transform_meta,
"DataTransformParam": transform_param
}
def load_model(self, model_meta, model_param):
self.delimitor, self.data_type, _0, self.tag_with_value, self.tag_value_delimitor, self.with_label, \
self.label_type, self.output_format, self.header, self.sid_name, self.label_name, self.with_match_id, \
self.anonymous_header = load_data_transform_model(
"SparseTagTransformer",
model_meta,
model_param)
self.missing_fill, self.missing_fill_method, \
self.missing_impute, self.default_value = load_missing_imputer_model(self.header,
"Imputer",
model_meta.imputer_meta,
model_param.imputer_param)
class DataTransform(ModelBase):
def __init__(self):
super(DataTransform, self).__init__()
self.transformer = None
from federatedml.param.data_transform_param import DataTransformParam
self.model_param = DataTransformParam()
self._input_model_meta = None
self._input_model_param = None
def _load_reader(self, schema=None):
if schema is None or not schema.get("meta", {}):
input_format = self.model_param.input_format
else:
input_format = schema["meta"].get("input_format")
if input_format == "dense":
self.transformer = DenseFeatureTransformer(self.model_param)
elif input_format == "sparse" or input_format == "svmlight":
self.transformer = SparseFeatureTransformer(self.model_param)
elif input_format == "tag":
self.transformer = SparseTagTransformer(self.model_param)
else:
raise ValueError("Cannot recognize input format")
if self._input_model_meta:
self.transformer.load_model(self._input_model_meta, self._input_model_param)
self._input_model_meta, self._input_model_param = None, None
self.transformer.anonymous_generator = Anonymous(self.role, self.component_properties.local_partyid)
def _init_model(self, model_param):
self.model_param = model_param
def load_model(self, model_dict):
for _, value in model_dict["model"].items():
for model in value:
if model.endswith("Meta"):
self._input_model_meta = value[model]
if model.endswith("Param"):
self._input_model_param = value[model]
def fit(self, data):
self._load_reader(data.schema)
data_inst = self.transformer.read_data(data, "fit")
if isinstance(self.transformer, (DenseFeatureTransformer, SparseTagTransformer)):
summary_buf = self.transformer.get_summary()
if summary_buf:
self.set_summary(summary_buf)
clear_schema(data_inst)
return data_inst
def transform(self, data):
self._load_reader(data.schema)
data_inst = self.transformer.read_data(data, "transform")
clear_schema(data_inst)
return data_inst
def export_model(self):
if not self.need_run:
model_meta = DataTransformMeta()
model_meta.need_run = False
model_param = DataTransformParam()
model_dict = dict(DataTransformMeta=model_param,
DataTransformParam=model_param)
else:
model_dict = self.transformer.save_model()
return model_dict
def clear_schema(data_inst):
ret_schema = copy.deepcopy(data_inst.schema)
key_words = {"sid", "header", "anonymous_header", "label_name",
"anonymous_label", "match_id_name"}
for key in data_inst.schema:
if key not in key_words:
del ret_schema[key]
data_inst.schema = ret_schema
def set_schema(data_instance, schema):
data_instance.schema = schema
def save_data_transform_model(input_format="dense",
delimitor=",",
data_type="str",
exclusive_data_type=None,
tag_with_value=False,
tag_value_delimitor=":",
with_label=False,
label_name='',
label_type="int",
output_format="dense",
header=None,
sid_name=None,
with_match_id=False,
model_name="DataTransform",
anonymous_header=None):
model_meta = DataTransformMeta()
model_param = DataTransformParam()
model_meta.input_format = input_format
model_meta.delimitor = delimitor
model_meta.data_type = data_type
model_meta.tag_with_value = tag_with_value
model_meta.tag_value_delimitor = tag_value_delimitor
model_meta.with_label = with_label
if with_label:
model_meta.label_name = label_name
model_meta.label_type = label_type
model_meta.output_format = output_format
model_meta.with_match_id = with_match_id
if header is not None:
model_param.header.extend(header)
if anonymous_header is not None:
model_param.anonymous_header.extend(anonymous_header)
if sid_name:
model_param.sid_name = sid_name
if label_name:
model_param.label_name = label_name
if exclusive_data_type is not None:
model_meta.exclusive_data_type.update(exclusive_data_type)
return model_meta, model_param
def load_data_transform_model(model_name="DataTransform",
model_meta=None,
model_param=None):
delimitor = model_meta.delimitor
data_type = model_meta.data_type
tag_with_value = model_meta.tag_with_value
tag_value_delimitor = model_meta.tag_value_delimitor
with_label = model_meta.with_label
label_name = model_meta.label_name if with_label else None
label_type = model_meta.label_type if with_label else None
try:
with_match_id = model_meta.with_match_id
except AttributeError:
with_match_id = False
output_format = model_meta.output_format
header = list(model_param.header) or None
try:
anonymous_header = list(model_param.anonymous_header)
except AttributeError:
anonymous_header = None
sid_name = None
if model_param.sid_name:
sid_name = model_param.sid_name
exclusive_data_type = None
if model_meta.exclusive_data_type:
exclusive_data_type = {}
for col_name in model_meta.exclusive_data_type:
exclusive_data_type[col_name] = model_meta.exclusive_data_type.get(col_name)
return delimitor, data_type, exclusive_data_type, tag_with_value, tag_value_delimitor, with_label, \
label_type, output_format, header, sid_name, label_name, with_match_id, anonymous_header
def save_missing_imputer_model(missing_fill=False,
missing_replace_method=None,
missing_impute=None,
missing_fill_value=None,
missing_replace_rate=None,
header=None,
model_name="Imputer"):
model_meta = DataTransformImputerMeta()
model_param = DataTransformImputerParam()
model_meta.is_imputer = missing_fill
if missing_fill:
if missing_replace_method:
model_meta.strategy = str(missing_replace_method)
if missing_impute is not None:
model_meta.missing_value.extend(map(str, missing_impute))
if missing_fill_value is not None:
feature_value_dict = dict(zip(header, map(str, missing_fill_value)))
model_param.missing_replace_value.update(feature_value_dict)
if missing_replace_rate is not None:
missing_replace_rate_dict = dict(zip(header, missing_replace_rate))
model_param.missing_value_ratio.update(missing_replace_rate_dict)
return model_meta, model_param
def load_missing_imputer_model(header=None,
model_name="Imputer",
model_meta=None,
model_param=None):
missing_fill = model_meta.is_imputer
missing_replace_method = model_meta.strategy
missing_value = model_meta.missing_value
missing_fill_value = model_param.missing_replace_value
if missing_fill:
if not missing_replace_method:
missing_replace_method = None
if not missing_value:
missing_value = None
else:
missing_value = list(missing_value)
if missing_fill_value:
missing_fill_value = [missing_fill_value.get(head) for head in header]
else:
missing_fill_value = None
else:
missing_replace_method = None
missing_value = None
missing_fill_value = None
return missing_fill, missing_replace_method, missing_value, missing_fill_value
def save_outlier_model(outlier_replace=False,
outlier_replace_method=None,
outlier_impute=None,
outlier_replace_value=None,
outlier_replace_rate=None,
header=None,
model_name="Outlier"):
model_meta = DataTransformOutlierMeta()
model_param = DataTransformOutlierParam()
model_meta.is_outlier = outlier_replace
if outlier_replace:
if outlier_replace_method:
model_meta.strategy = str(outlier_replace_method)
if outlier_impute:
model_meta.outlier_value.extend(map(str, outlier_impute))
if outlier_replace_value:
outlier_value_dict = dict(zip(header, map(str, outlier_replace_value)))
model_param.outlier_replace_value.update(outlier_value_dict)
if outlier_replace_rate:
outlier_value_ratio_dict = dict(zip(header, outlier_replace_rate))
model_param.outlier_value_ratio.update(outlier_value_ratio_dict)
return model_meta, model_param
def load_outlier_model(header=None,
model_name="Outlier",
model_meta=None,
model_param=None):
outlier_replace = model_meta.is_outlier
outlier_replace_method = model_meta.strategy
outlier_value = model_meta.outlier_value
outlier_replace_value = model_param.outlier_replace_value
if outlier_replace:
if not outlier_replace_method:
outlier_replace_method = None
if not outlier_value:
outlier_value = None
else:
outlier_value = list(outlier_value)
if outlier_replace_value:
outlier_replace_value = [outlier_replace_value.get(head) for head in header]
else:
outlier_replace_value = None
else:
outlier_replace_method = None
outlier_value = None
outlier_replace_value = None
return outlier_replace, outlier_replace_method, outlier_value, outlier_replace_value