Removing Outliers with Standard Deviation Bounds in Spark

1.The rule: average plus or minus k standard deviations

Every way of removing outliers from a Spark DataFrame follows the same pattern. You reduce a numeric column to a lower and an upper bound, then keep the rows that fall inside. The standard deviation method takes those bounds from two statistics of the column, its average and its standard deviation. It steps a fixed number of standard deviations to either side of the average.

The Databricks data-quality guidance documents this pattern and uses it to find outliers and anomalies. The bounds come from one aggregate query over the reference data:

bounds = spark.sql("""
    SELECT
      avg(amount) - 3 * stddev(amount) as lower_bound,
      avg(amount) + 3 * stddev(amount) as upper_bound
    FROM historical_stats
    WHERE
      date >= CURRENT_DATE() - INTERVAL 30 DAYS
  """)

  # Join with new data and apply bounds
  return spark.read.table("new_data").crossJoin(bounds)

Look at the two expressions. lower_bound is the average minus three standard deviations, and upper_bound is the average plus three. The multiplier 3 is a choice this example makes, not something Spark fixes. It is a parameter of the method, and a smaller multiplier pulls the bounds closer to the average, so more rows end up outside them.

The bounds also come from a different table than the one they are applied to. They are computed over historical_stats and applied to new_data through crossJoin. The aggregate has no grouping, so it returns a single row. Crossing that row with the data adds lower_bound and upper_bound to every row, ready to be compared.

Sources1

2.Computing the mean and standard deviation in PySpark

In the DataFrame API, both statistics come from pyspark.sql.functions. sf.stddev is an alias for stddev_samp. The SQL stddev function is documented the same way: it returns the sample standard deviation of the values in the group. sf.mean returns the average and is an alias of avg. Both names produce the same result, and the output header reads avg(...) either way.

import pyspark.sql.functions as sf
spark.range(6).select(sf.stddev("id")).show()

+------------------+
|        stddev(id)|
+------------------+
|1.8708286933869...|
+------------------+

Three behaviours matter when you use these functions in an outlier rule.

First, they return a Column, not a number. They are expressions, so they have to be evaluated inside select or agg. df.agg aggregates the entire DataFrame without groups. That gives you the shape you need: one row holding the column's mean and standard deviation, which you can cross-join onto the data as shown above.

Second, a standard deviation needs more than one value. If any group consists of only one row, stddev returns NULL for that group.

Third, the SQL function accepts DISTINCT, which computes over unique values only. On the values 1, 2, 3, 3, stddev(col) returns 0.957…, but stddev(DISTINCT col) returns 1.0. Adding DISTINCT therefore changes the bounds you get.

import pyspark.sql.functions as sf
spark.range(6).select(sf. ? ("id")).show()

Sources234

3.Dropping the rows outside the bounds

Once you have the bounds, removing outliers is just a filter. df.filter takes a Column of BooleanType or a string of SQL expressions. It returns a new DataFrame containing the rows that satisfy the condition. Because the condition describes the rows you keep, not the ones you drop, you write it for the inliers. Several comparisons can be combined with &:

df.filter((df.age > 3) & (df.subject == "Physics")).show()
# +---+----+-------+
# |age|name|subject|
# +---+----+-------+
# |  5| Bob|Physics|
# +---+----+-------+

You could write the range as two comparisons, but Column.between expresses it directly. It checks whether the value lies between the lower and upper bounds, inclusive. Spark even renders the test as (age >= 2) AND (age <= 4):

df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], ["age", "name"])
df.select(df.name, df.age.between(2, 4)).show()

# +-----+---------------------------+
# | name|((age >= 2) AND (age <= 4))|
# +-----+---------------------------+
# |Alice|                       true|
# |  Bob|                      false|
# +-----+---------------------------+

Each bound can be a value or a Column. After the cross join, every row carries its own lower_bound and upper_bound columns, and between can compare against them directly. The Databricks example writes the same test in SQL as amount BETWEEN lower_bound AND upper_bound.

To see the outliers instead of the inliers, negate the test. NOT BETWEEN is a synonym for expr1 < expr2 or expr1 > expr3, so it lets you look at the rows before you throw them away. Watch out for a missing bound: 4 not between NULL and 5 evaluates to NULL, not true or false.

Sources561

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.sf.stddev returns the population standard deviation.Why is that wrong?

   sf.stddev is an alias for stddev_samp, so it returns the sample standard deviation.

   Covered in Computing the mean and standard deviation in PySpark

2. 2.A standard deviation over a single row is 0, so the bounds collapse onto that one value.Why is that wrong?

   stddev returns NULL for any group that has only one row, so the bounds built from it are NULL as well.

   Covered in Computing the mean and standard deviation in PySpark

3. 3.Column.between excludes the bound values, so a row sitting exactly on a bound is removed.Why is that wrong?

   between is inclusive at both ends and is equivalent to >= AND <=, so a row equal to a bound is kept.

   Covered in Dropping the rows outside the bounds