Optimized variance in Cascalog

gistfile1.clj

(defn one [] 1)

(defn div [v1 v2]
  (float (/ v1 v2)))

(defparallelagg count
  :init-var #'one
  :combine-var #'+)

(defparallelagg sum-parallel
  :init-var #'identity
  :combine-var #'+)

(def sum (each sum-parallel))

(def avg
  (<- [!val :> !avg]
    (count !count)
    (sum !val :> !sum)
    (div !sum !count :> !avg)))

(def variance
  (<- [!val :> !var]
    (* !val !val :> !squared)
    (sum !squared :> !square-sum)
    (count !count)
    (avg !val :> !mean)
    (* !mean !mean :> !mean-squared)
    (div !square-sum !count :> !i)
    (- !i !mean-squared :> !var)))