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