<s> import bisect <EOL> from dpark . util import portable_hash <EOL> from dpark . serialize import load_func , dump_func <EOL> class Dependency : <EOL> def __init__ ( self , rdd ) : <EOL> self . rdd = rdd <EOL> def __getstate__ ( self ) : <EOL> raise ValueError ( "<STR_LIT>" % self ) <EOL> class NarrowDependency ( Dependency ) : <EOL> isShuffle = False <EOL> def getParents ( self , outputPartition ) : <EOL> raise NotImplementedError <EOL> class OneToOneDependency ( NarrowDependency ) : <EOL> def getParents ( self , pid ) : <EOL> return [ pid ] <EOL> class OneToRangeDependency ( NarrowDependency ) : <EOL> def __init__ ( self , rdd , splitSize , length ) : <EOL> Dependency . __init__ ( self , rdd ) <EOL> self . splitSize = splitSize <EOL> self . length = length <EOL> def getParents ( self , pid ) : <EOL> return range ( pid * self . splitSize , <EOL> min ( ( pid + <NUM_LIT:1> ) * self . splitSize , self . length ) ) <EOL> class CartesianDependency ( NarrowDependency ) : <EOL> def __init__ ( self , rdd , first , numSplitsInRdd2 ) : <EOL> NarrowDependency . __init__ ( self , rdd ) <EOL> self . first = first <EOL> self . numSplitsInRdd2 = numSplitsInRdd2 <EOL> def getParents ( self , pid ) : <EOL> if self . first : <EOL> return [ pid / self . numSplitsInRdd2 ] <EOL> else : <EOL> return [ pid % self . numSplitsInRdd2 ] <EOL> class RangeDependency ( NarrowDependency ) : <EOL> def __init__ ( self , rdd , inStart , outStart , length ) : <EOL> Dependency . __init__ ( self , rdd ) <EOL> self . inStart = inStart <EOL> self . outStart = outStart <EOL> self . length = length <EOL> def getParents ( self , pid ) : <EOL> if pid >= self . outStart and pid < self . outStart + self . length : <EOL> return [ pid - self . outStart + self . inStart ] <EOL> return [ ] <EOL> class ShuffleDependency ( Dependency ) : <EOL> isShuffle = True <EOL> def __init__ ( self , shuffleId , rdd , aggregator , partitioner ) : <EOL> Dependency . __init__ ( self , rdd ) <EOL> self . shuffleId = shuffleId <EOL> self . aggregator = aggregator <EOL> self . partitioner = partitioner <EOL> class Aggregator : <EOL> def __init__ ( self , createCombiner , mergeValue , <EOL> mergeCombiners ) : <EOL> self . createCombiner = createCombiner <EOL> self . mergeValue = mergeValue <EOL> self . mergeCombiners = mergeCombiners <EOL> def __getstate__ ( self ) : <EOL> return ( dump_func ( self . createCombiner ) , <EOL> dump_func ( self . mergeValue ) , <EOL> dump_func ( self . mergeCombiners ) ) <EOL> def __setstate__ ( self , state ) : <EOL> c1 , c2 , c3 = state <EOL> self . createCombiner = load_func ( c1 ) <EOL> self . mergeValue = load_func ( c2 ) <EOL> self . mergeCombiners = load_func ( c3 ) <EOL> class AddAggregator : <EOL> def createCombiner ( self , x ) : <EOL> return x <EOL> def mergeValue ( self , s , x ) : <EOL> return s + x <EOL> def mergeCombiners ( self , x , y ) : <EOL> return x + y <EOL> class MergeAggregator : <EOL> def createCombiner ( self , x ) : <EOL> return [ x ] <EOL> def mergeValue ( self , s , x ) : <EOL> s . append ( x ) <EOL> return s <EOL> def mergeCombiners ( self , x , y ) : <EOL> x . extend ( y ) <EOL> return x <EOL> class UniqAggregator : <EOL> def createCombiner ( self , x ) : <EOL> return set ( [ x ] ) <EOL> def mergeValue ( self , s , x ) : <EOL> s . add ( x ) <EOL> return s <EOL> def mergeCombiners ( self , x , y ) : <EOL> x |= y <EOL> return x <EOL> class Partitioner : <EOL> @ property <EOL> def numPartitions ( self ) : <EOL> raise NotImplementedError <EOL> def getPartition ( self , key ) : <EOL> raise NotImplementedError <EOL> class HashPartitioner ( Partitioner ) : <EOL> def __init__ ( self , partitions ) : <EOL> self . partitions = max ( <NUM_LIT:1> , int ( partitions ) ) <EOL> @ property <EOL> def numPartitions ( self ) : <EOL> return self . partitions <EOL> def getPartition ( self , key ) : <EOL> return portable_hash ( key ) % self . partitions <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , Partitioner ) : <EOL> return other . numPartitions == self . numPartitions <EOL> return False <EOL> class RangePartitioner ( Partitioner ) : <EOL> def __init__ ( self , keys , reverse = False ) : <EOL> self . keys = sorted ( keys ) <EOL> self . reverse = reverse <EOL> @ property <EOL> def numPartitions ( self ) : <EOL> return len ( self . keys ) + <NUM_LIT:1> <EOL> def getPartition ( self , key ) : <EOL> idx = bisect . bisect ( self . keys , key ) <EOL> return len ( self . keys ) - idx if self . reverse else idx <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , RangePartitioner ) : <EOL> return other . keys == self . keys and self . reverse == other . reverse <EOL> return False </s>
