<s> from operator import add <EOL> import copy <EOL> from dpark . serialize import load_func , dump_func <EOL> class AccumulatorParam : <EOL> def __init__ ( self , zero , addInPlace ) : <EOL> self . zero = zero <EOL> self . addInPlace = addInPlace <EOL> def __getstate__ ( self ) : <EOL> return dump_func ( self . addInPlace ) , self . zero <EOL> def __setstate__ ( self , state ) : <EOL> add , self . zero = state <EOL> self . addInPlace = load_func ( add ) <EOL> numAcc = AccumulatorParam ( <NUM_LIT:0> , add ) <EOL> listAcc = AccumulatorParam ( [ ] , lambda x , y : x . extend ( y ) or x ) <EOL> mapAcc = AccumulatorParam ( { } , lambda x , y : x . update ( y ) or x ) <EOL> setAcc = AccumulatorParam ( set ( ) , lambda x , y : x . update ( y ) or x ) <EOL> class Accumulator : <EOL> def __init__ ( self , initialValue = <NUM_LIT:0> , param = numAcc ) : <EOL> self . id = self . newId ( ) <EOL> if param is None : <EOL> param = numAcc <EOL> self . param = param <EOL> self . value = initialValue <EOL> self . register ( self , True ) <EOL> def add ( self , v ) : <EOL> self . value = self . param . addInPlace ( self . value , v ) <EOL> self . register ( self , False ) <EOL> def reset ( self ) : <EOL> v = self . value <EOL> self . value = copy . copy ( self . param . zero ) <EOL> return v <EOL> def __getstate__ ( self ) : <EOL> return self . id , self . param <EOL> def __setstate__ ( self , s ) : <EOL> self . id , self . param = s <EOL> self . value = copy . copy ( self . param . zero ) <EOL> self . register ( self , False ) <EOL> nextId = <NUM_LIT:0> <EOL> @ classmethod <EOL> def newId ( cls ) : <EOL> cls . nextId += <NUM_LIT:1> <EOL> return cls . nextId <EOL> originals = { } <EOL> localAccums = { } <EOL> @ classmethod <EOL> def register ( cls , acc , original ) : <EOL> if original : <EOL> cls . originals [ acc . id ] = acc <EOL> else : <EOL> cls . localAccums [ acc . id ] = acc <EOL> @ classmethod <EOL> def clear ( cls ) : <EOL> for acc in cls . localAccums . values ( ) : <EOL> acc . reset ( ) <EOL> cls . localAccums . clear ( ) <EOL> @ classmethod <EOL> def values ( cls ) : <EOL> v = dict ( ( id , accum . value ) for id , accum in cls . localAccums . items ( ) ) <EOL> cls . clear ( ) <EOL> return v <EOL> @ classmethod <EOL> def merge ( cls , values ) : <EOL> for id , value in values . items ( ) : <EOL> cls . originals [ id ] . add ( value ) <EOL> ReadBytes = Accumulator ( ) <EOL> WriteBytes = Accumulator ( ) <EOL> RemoteReadBytes = Accumulator ( ) <EOL> LocalReadBytes = Accumulator ( ) <EOL> CacheHits = Accumulator ( ) <EOL> CacheMisses = Accumulator ( ) </s>
