[IDEA] Full pipeline with Custom Transformer and feature names for Chapter 2
- Dominant language
- Jupyter Notebook
- Stars
- 30k
- Forks
- 13.1k
- PR merge metrics
- No merged PRs in 30d
Description
Given this pipeline
```
num_cols = ['longitude', 'latitude', 'housing_median_age', 'total_rooms',
'total_bedrooms', 'population', 'households', 'median_income']
cat_cols = ['ocean_proximity']
num_transformer = Pipeline(steps=[
('impute', SimpleImputer(strategy='median')),
('add_feats', CombinedAttributesAdder(add_bedrooms_per_room=True)),
('scaler', StandardScaler())
])
cat_transformer = Pipeline(steps=[
('impute', SimpleImputer(strategy='most_frequent')),
('ohe', OneHotEncoder(handle_unknown='ignore'))
])
preprocessor = ColumnTransformer(transformers=[
('numeric', num_transformer, num_cols),
('categorical', cat_transformer, cat_cols)
])
```
I wanted to be able to inspect the final transformed data as a DataFrame
```
housing_tr = pd.DataFrame(preprocessor.transform(housing),
columns=preprocessor.get_feature_names_out(), # this didn't work for me
index=housing.index)
```
This is what I've done to make it work
1. added get_feature_names_out method to CombinedAttributesAdder class
2. created custom function to get the feature names out from ColumnTransformer
**Added get_feature_names_out method to CombinedAttributesAdder class**
```
class CombinedAttributesAdder(BaseEstimator, TransformerMixin):
def __init__(self, add_bedrooms_per_room=True):
self.add_bedrooms_per_room = add_bedrooms_per_room
def fit(self, X, y=None):
self.n_features_in_ = X.shape[1] # added but not necessary
return self
def transform(self, X):
rix, bix, pix, hix = [num_cols.index(e) for e in ['total_rooms', 'total_bedrooms', 'population', 'households']]
rooms_per_household = X[:, rix] / X[:, hix]
population_per_household = X[:, pix] / X[:, hix]
if self.add_bedrooms_per_room:
bedrooms_per_room = X[:, bix] / X[:, rix]
return np.c_[X, rooms_per_household, population_per_household,
bedrooms_per_room]
else:
return np.c_[X, rooms_per_household, population_per_household]
def get_feature_names_out(self, input_features=None):
feature_names = input_features + ['rooms_per_household', 'population_per_household']
if self.add_bedrooms_per_room:
feature_names.extend(['bedrooms_per_room'])
return np.asarray(feature_names, dtype=object)
```
**Created custom function to get the feature names out from ColumnTransformer**
Note: this will probably only work for a limited configurations of ColumnTransformer's and similar to the aforementioned one.
```
def get_names_out_from_ColumnTransformer(column_transformer, df=None):
"""
Returns a list of the feature names produced by a Column Transformer
It should probably do the same as the ColumnTransformer method get_feature_names_out()
but it didn't work for me
column_transformer: an instance of a fitted Column Transformer
df: the dataframe passed to the fit method of a Column Transformer
(only needed if remainder step has passthrough strategy)
"""
from sklearn.utils.validation import check_is_fitted
check_is_fitted(column_transformer)
col_names = []
# column_transformer.transformers_ is a list of tuples
# each tuple (outer_pipeline) has three elements:
# name of the pipeline -> outer_pipeline[0]
# the fitted pipeline -> outer_pipeline[1]
# list of features fed into the pipeline -> outer_pipeline[2]
for outer_pipeline in [p for p in column_transformer.transformers_ if p[0] != 'remainder']:
features_in = outer_pipeline[2]
print(f"features in '{outer_pipeline[0]}': {features_in}")
for inner_pipeline_step in outer_pipeline[1].steps:
print(f" features in '{inner_pipeline_step[0]}': {features_in}")
# inner_pipeline_step is a tuple of two elements
# name of the transformer -> inner_pipeline_step[0]
# the fitted transformer -> inner_pipeline_step[1]
transformer = inner_pipeline_step[1]
if hasattr(transformer, 'get_feature_names_out'):
features_out = transformer.get_feature_names_out(features_in).tolist()
else:
# if a transformer doesn't have get_feature_names_out method
# features in = features out
features_out = features_in
# if an imputer has add_indicator=True make a name for it
if hasattr(transformer, 'indicator_') \
and transformer.indicator_ is not None:
features_out += [features_in[i] + '_missing' for i in transformer.indicator_.features_]
# features_out is features_in for the next inner_pipeline_step
features_in = features_out
print(f" features out '{inner_pipeline_step[0]}': {features_in}")
col_names.extend(features_out)
# add passthrough-ed columns
if 'remainder' in column_transformer.named_transformers_.keys() \
and column_transformer.named_transformers_['remainder'] == 'passthrough':
assert df is not None, "df is None"
remainder = column_transformer.transformers_[-1]
passthrough_features = df.columns[remainder[2]].tolist()
col_names.extend(passthrough_features)
return col_names
```
The below should work fine now
```
housing_tr = pd.DataFrame(preprocessor.transform(housing),
columns=get_names_out_from_ColumnTransformer(preprocessor, housing),
index=housing.index)
```
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.