Normalize values of multiple columns in Spark DataFrame, using only DataFrame API

Question

I am trying to normalize the values of multiple columns in a spark dataframe, by subtracting the mean and dividing by the stddev of each column. Here's the code I have so far:

from pyspark.sql import Row
from pyspark.sql.functions import stddev_pop, avg

df = spark.createDataFrame([Row(A=1, B=6), Row(A=2, B=7), Row(A=3, B=8),
                            Row(A=4, B=9), Row(A=5, B=10)])

exprs = [x - (avg(x)) / stddev_pop(x) for x in df.columns]    
df.select(exprs).show()

Which gives me the result:

+------------------------------+------------------------------+
|(A - (avg(A) / stddev_pop(A)))|(B - (avg(B) / stddev_pop(B)))|
+------------------------------+------------------------------+
|                          null|                          null|
+------------------------------+------------------------------+

Where I'm hoping for:

+------------------------------+------------------------------+
|(A - (avg(A) / stddev_pop(A)))|(B - (avg(B) / stddev_pop(B)))|
+------------------------------+------------------------------+
|                  -1.414213562|                  -1.414213562|
|                  -0.707106781|                  -0.707106781|
|                             0|                             0|
|                   0.707106781|                   0.707106781|
|                   1.414213562|                   1.414213562|
+------------------------------+------------------------------+

I believe I can do this with the StandardScaler class from mllib, but I'd prefer to do this using only the dataframe API if possible - if only as a learning exercise.

Possible duplicate of How to create z score in spark sql for each group — user6022341
– user6022341, Commented Oct 12, 2016 at 11:43
@LostInOverflow it's certainly close, but my use case required normalising multiple columns — MarkNS
– MarkNS, Commented Oct 12, 2016 at 11:47

Community · Accepted Answer · 2017-05-23 12:19:30Z

6

With thanks to the answer here, I came up with this:

from pyspark.sql.functions import stddev_pop, avg, broadcast

cols = df.columns    
stats = (df.groupBy().agg(
        *([stddev_pop(x).alias(x + '_stddev') for x in cols] + 
          [avg(x).alias(x + '_avg') for x in cols])))

df = df.join(broadcast(stats))

exprs = [(df[x] - df[x + '_avg']) / df[x + '_stddev'] for x in cols]
df.select(exprs).show()

+------------------------+------------------------+
|((A - A_avg) / A_stddev)|((B - B_avg) / B_stddev)|
+------------------------+------------------------+
|      -1.414213562373095|      -1.414213562373095|
|     -0.7071067811865475|     -0.7071067811865475|
|                     0.0|                     0.0|
|      0.7071067811865475|      0.7071067811865475|
|       1.414213562373095|       1.414213562373095|
+------------------------+------------------------+

edited May 23, 2017 at 12:19

CommunityBot

11 silver badge

answered Oct 12, 2016 at 8:47

MarkNS

4,1414 gold badges48 silver badges68 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

rosefun Over a year ago

Why I meet error Join condition is missing or trivial. in spark 2.3

Collectives™ on Stack Overflow

Normalize values of multiple columns in Spark DataFrame, using only DataFrame API

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related