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          self.maxValue = float("-inf") 
 30          self.minValue = float("inf") 
 31           
 32          for v in values: 
 33              self.merge(v) 
 34               
 35      # Add a value into this StatCounter, updating the internal statistics. 
 36 -    def merge(self, value): 
 37          delta = value - self.mu 
 38          self.n += 1 
 39          self.mu += delta / self.n 
 40          self.m2 += delta * (value - self.mu) 
 41          if self.maxValue < value: 
 42              self.maxValue = value 
 43          if self.minValue > value: 
 44              self.minValue = value 
 45               
 46          return self 
 47   
 48      # Merge another StatCounter into this one, adding up the internal statistics. 
 49 -    def mergeStats(self, other): 
 50          if not isinstance(other, StatCounter): 
 51              raise Exception("Can only merge Statcounters!") 
 52   
 53          if other is self: # reference equality holds 
 54              self.merge(copy.deepcopy(other))  # Avoid overwriting fields in a weird order 
 55          else: 
 56              if self.n == 0: 
 57                  self.mu = other.mu 
 58                  self.m2 = other.m2 
 59                  self.n = other.n 
 60                  self.maxValue = other.maxValue 
 61                  self.minValue = other.minValue 
 62                   
 63              elif other.n != 0:         
 64                  delta = other.mu - self.mu 
 65                  if other.n * 10 < self.n: 
 66                      self.mu = self.mu + (delta * other.n) / (self.n + other.n) 
 67                  elif self.n * 10 < other.n: 
 68                      self.mu = other.mu - (delta * self.n) / (self.n + other.n) 
 69                  else: 
 70                      self.mu = (self.mu * self.n + other.mu * other.n) / (self.n + other.n) 
 71                   
 72                      self.maxValue = max(self.maxValue, other.maxValue) 
 73                      self.minValue = min(self.minValue, other.minValue) 
 74           
 75                  self.m2 += other.m2 + (delta * delta * self.n * other.n) / (self.n + other.n) 
 76                  self.n += other.n 
 77          return self 
 78   
 79      # Clone this StatCounter 
 80 -    def copy(self): 
 81          return copy.deepcopy(self) 
 82   
 83 -    def count(self): 
 84          return self.n 
 85   
 86 -    def mean(self): 
 87          return self.mu 
 88   
 89 -    def sum(self): 
 90          return self.n * self.mu 
 91   
 92 -    def min(self): 
 93          return self.minValue 
 94   
 95 -    def max(self): 
 96          return self.maxValue 
 97       
 98      # Return the variance of the values. 
 99 -    def variance(self): 
100          if self.n == 0: 
101              return float('nan') 
102          else: 
103              return self.m2 / self.n 
104   
105      # 
106      # Return the sample variance, which corrects for bias in estimating the variance by dividing 
107      # by N-1 instead of N. 
108      # 
109 -    def sampleVariance(self): 
110          if self.n <= 1: 
111              return float('nan') 
112          else: 
113              return self.m2 / (self.n - 1) 
114   
115      # Return the standard deviation of the values. 
116 -    def stdev(self): 
117          return math.sqrt(self.variance()) 
118   
119      # 
120      # Return the sample standard deviation of the values, which corrects for bias in estimating the 
121      # variance by dividing by N-1 instead of N. 
122      # 
123 -    def sampleStdev(self): 
124          return math.sqrt(self.sampleVariance()) 
125   
126 -    def __repr__(self): 
127          return "(count: %s, mean: %s, stdev: %s, max: %s, min: %s)" % (self.count(), self.mean(), self.stdev(), self.max(), self.min()) 
128