code: 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();
  			}
  		};
  	}
docs: Sample algorithm for the second phase. This operation should be executed as the UDF of
  an all reduce operation.
  
  @param input The intermediate sample output generated in the first phase.
  @return The sampled output.