public Iterator < T > sampleInCoordinator ( Iterator < IntermediateSampleData < T > > input ) { if ( numSamples == 0 ) { return emptyIterable ; } // This queue holds fixed number elements with the top K weight for the coordinator. PriorityQueue < IntermediateSampleData < T > > reservoir = new PriorityQueue < IntermediateSampleData < T > > ( numSamples ) ; int index = 0 ; IntermediateSampleData < T > smallest = null ; while ( input . hasNext ( ) ) { IntermediateSampleData < T > element = input . next ( ) ; if ( index < numSamples ) { // Fill the queue with first K elements from input. reservoir . add ( element ) ; smallest = reservoir . peek ( ) ; } else { // If current element weight is larger than the smallest one in queue, remove the element // with the smallest weight, and append current element into the queue. if ( element . getWeight ( ) > smallest . getWeight ( ) ) { reservoir . remove ( ) ; reservoir . add ( element ) ; smallest = reservoir . peek ( ) ; } } index ++ ; } final Iterator < IntermediateSampleData < T > > itr = reservoir . iterator ( ) ; return new Iterator < T > ( ) { @ Override public boolean hasNext ( ) { return itr . hasNext ( ) ; } @ Override public T next ( ) { return itr . next ( ) . getElement ( ) ; } @ Override public void remove ( ) { itr . remove ( ) ; } } ; }
Sample algorithm for the second phase . This operation should be executed as the UDF of an all reduce operation .