I have the below data frame and I want to remove outliers from defined columns. In the below example price and income. Outliers should be removed for each group of data. In this example its 'cd' and 'segment' columns. Outliers should be removed based 5 standard deviations.
data = [
('a', '1',20,10),
('a', '1',30,16),
('a', '1',50,91),
('a', '1',60,34),
('a', '1',200,23),
('a', '2',33,87),
('a', '2',86,90),
('a','2',89,35),
('a', '2',90,24),
('a', '2',40,97),
('a', '2',1,21),
('b', '1',45,96),
('b', '1',56,99),
('b', '1',89,23),
('b', '1',98,64),
('b', '2',86,42),
('b', '2',45,54),
('b', '2',67,95),
('b','2',86,70),
('b', '2',91,64),
('b', '2',2,53),
('b', '2',4,87)
]
data = (spark.createDataFrame(data, ['cd','segment','price','income']))
I have used the code below to remove outliers but this would work only for one column.
mean_std = (
data
.groupBy('cd', 'segment')
.agg(
*[f.mean(colName).alias('{}{}'.format('mean_',colName)) for colName in ['price']],
*[f.stddev(colName).alias('{}{}'.format('stddev_',colName)) for colName in ['price']])
)
mean_columns = ['mean_price']
std_columns = ['stddev_price']
upper = mean_std
for col_1 in mean_columns:
for col_2 in std_columns:
if col_1 != col_2:
name = col_1 + '_upper_limit'
upper = upper.withColumn(name, f.col(col_1) + f.col(col_2)*5)
lower = upper
for col_1 in mean_columns:
for col_2 in std_columns:
if col_1 != col_2:
name = col_1 + '_lower_limit'
lower = lower.withColumn(name, f.col(col_1) - f.col(col_2)*5)
outliers = (data.join(lower,
how = 'left',
on = ['cd', 'segment'])
.withColumn('is_outlier_price', f.when((f.col('price')>f.col('mean_price_upper_limit')) |
(f.col('price')<f.col('mean_price_lower_limit')),1)
.otherwise(None))
)
my final output should have a column for each variable stating whether its 1 = remove or 0 = keep.
Really appreciate any help on this.