title: PySpark: calculate mean, standard deviation and those values around the mean in one step
body: <p>My raw data comes in a tabular format. It contains observations from different variables. Each observation with the variable name, the timestamp and the value at that time.</p>
  
  <blockquote>
    <p>Variable [string], Time [datetime], Value [float]</p>
  </blockquote>
  
  <p>The data is stored as Parquet in HDFS and loaded into a Spark Dataframe (df). From that dataframe.</p>
  
  <p>Now I want to calculate default statistics like Mean, Standard Deviation and others for each variable. Afterwards, once the Mean has been retrieved, I want to filter/count those values for that variable that are closely around the Mean.</p>
  
  <p>Due to the answer towards my <a href="https://stackoverflow.com/questions/38227007/pyspark-retrieve-mean-and-the-count-of-values-around-the-mean-for-groups-within/38246827?noredirect=1#comment63917723_38246827">other question</a>, I came up with this code:</p>
  
  <pre><code>from pyspark.sql.window import Window
  from pyspark.sql.functions import *
  from pyspark.sql.types import *
  
  w1 = Window().partitionBy("Variable")
  w2 = Window.partitionBy("Variable").orderBy("Time")
  
  def stddev_pop_w(col, w):
      #Built-in stddev doesn't support windowing
      return sqrt(avg(col * col).over(w) - pow(avg(col).over(w), 2))
  
  def isInRange(value, mean, stddev, radius):
      try:
          if (abs(value - mean) &lt; radius * stddev):
              return 1
          else:
              return 0
      except AttributeError:
          return -1
  
  delta = col("Time").cast("long") - lag("Time", 1).over(w2).cast("long")
  #f = udf(lambda (value, mean, stddev, radius): abs(value - mean) &lt; radius * stddev, IntegerType())
  #f2 = udf(lambda value, mean, stddev: isInRange(value, mean, stddev, 2), IntegerType())
  #f3 = udf(lambda value, mean, stddev: isInRange(value, mean, stddev, 3), IntegerType())
  
  df_ = df_all \
      .withColumn("mean", mean("Value").over(w1)) \
      .withColumn("std_deviation", stddev_pop_w(col("Value"), w1)) \
      .withColumn("delta", delta) \
  #    .withColumn("stddev_2", f2("Value", "mean", "std_deviation")) \
  #    .withColumn("stddev_3", f3("Value", "mean", "std_deviation")) \
  
  #df2.show(5, False)
  </code></pre>
  
  <p><strong>Question:</strong> The last two commented-lines won't work. It will give an AttributeError because the incoming values for stddev and mean are null. I guess this happens because I'm referring to columns that are also just calculated on the fly and have no value at that moment. <strong>But is there a way to achieve that?</strong></p>
  
  <p>Currently I'm doing a second run like this:</p>
  
  <pre><code>df = df_.select("*", \
      abs(df_.Value - df_.mean).alias("max_deviation_mean"), \
      when(abs(df_.Value - df_.mean) &lt; 2 * df_.std_deviation, 1).otherwise(1).alias("std_dev_mean_2"), \
      when(abs(df_.Value - df_.mean) &lt; 3 * df_.std_deviation, 1).otherwise(1).alias("std_dev_mean_3"))
  </code></pre>