Source Code for Module pyspark.statcounter

  1  # 
  2  # Licensed to the Apache Software Foundation (ASF) under one or more 
  3  # contributor license agreements.  See the NOTICE file distributed with 
  4  # this work for additional information regarding copyright ownership. 
  5  # The ASF licenses this file to You under the Apache License, Version 2.0 
  6  # (the "License"); you may not use this file except in compliance with 
  7  # the License.  You may obtain a copy of the License at 
  8  # 
  9  #    http://www.apache.org/licenses/LICENSE-2.0 
 10  # 
 11  # Unless required by applicable law or agreed to in writing, software 
 12  # distributed under the License is distributed on an "AS IS" BASIS, 
 13  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 14  # See the License for the specific language governing permissions and 
 15  # limitations under the License. 
 16  # 
 17   
 18  # This file is ported from spark/util/StatCounter.scala 
 19   
 20  import copy 
 21  import math 
 22   
 23 -class StatCounter(object): 
 24       
 25 -    def __init__(self, values=[]): 
 26          self.n = 0L    # Running count of our values 
 27          self.mu = 0.0  # Running mean of our values 
 28          self.m2 = 0.0  # Running variance numerator (sum of (x - mean)^2) 
 29   
 30          for v in values: 
 31              self.merge(v) 
 32               
 33      # Add a value into this StatCounter, updating the internal statistics. 
 34 -    def merge(self, value): 
 35          delta = value - self.mu 
 36          self.n += 1 
 37          self.mu += delta / self.n 
 38          self.m2 += delta * (value - self.mu) 
 39          return self 
 40   
 41      # Merge another StatCounter into this one, adding up the internal statistics. 
 42 -    def mergeStats(self, other): 
 43          if not isinstance(other, StatCounter): 
 44              raise Exception("Can only merge Statcounters!") 
 45   
 46          if other is self: # reference equality holds 
 47              self.merge(copy.deepcopy(other))  # Avoid overwriting fields in a weird order 
 48          else: 
 49              if self.n == 0: 
 50                  self.mu = other.mu 
 51                  self.m2 = other.m2 
 52                  self.n = other.n        
 53              elif other.n != 0:         
 54                  delta = other.mu - self.mu 
 55                  if other.n * 10 < self.n: 
 56                      self.mu = self.mu + (delta * other.n) / (self.n + other.n) 
 57                  elif self.n * 10 < other.n: 
 58                      self.mu = other.mu - (delta * self.n) / (self.n + other.n) 
 59                  else: 
 60                      self.mu = (self.mu * self.n + other.mu * other.n) / (self.n + other.n) 
 61           
 62                  self.m2 += other.m2 + (delta * delta * self.n * other.n) / (self.n + other.n) 
 63                  self.n += other.n 
 64          return self 
 65   
 66      # Clone this StatCounter 
 67 -    def copy(self): 
 68          return copy.deepcopy(self) 
 69   
 70 -    def count(self): 
 71          return self.n 
 72   
 73 -    def mean(self): 
 74          return self.mu 
 75   
 76 -    def sum(self): 
 77          return self.n * self.mu 
 78   
 79      # Return the variance of the values. 
 80 -    def variance(self): 
 81          if self.n == 0: 
 82              return float('nan') 
 83          else: 
 84              return self.m2 / self.n 
 85   
 86      # 
 87      # Return the sample variance, which corrects for bias in estimating the variance by dividing 
 88      # by N-1 instead of N. 
 89      # 
 90 -    def sampleVariance(self): 
 91          if self.n <= 1: 
 92              return float('nan') 
 93          else: 
 94              return self.m2 / (self.n - 1) 
 95   
 96      # Return the standard deviation of the values. 
 97 -    def stdev(self): 
 98          return math.sqrt(self.variance()) 
 99   
100      # 
101      # Return the sample standard deviation of the values, which corrects for bias in estimating the 
102      # variance by dividing by N-1 instead of N. 
103      # 
104 -    def sampleStdev(self): 
105          return math.sqrt(self.sampleVariance()) 
106   
107 -    def __repr__(self): 
108          return "(count: %s, mean: %s, stdev: %s)" % (self.count(), self.mean(), self.stdev()) 
109