Reducer class

KMeansReduce.java

public class KMeansReducer extends Reducer<PointVector, PointVector, Text, Text> {
    private double min_dist = Double.MAX_VALUE;

    @Override
    public void reduce(PointVector center, Iterable<PointVector> points, Context context) throws IOException, InterruptedException {
        EuclideanDistance measure = new EuclideanDistance();
        double distance = 0.0;
        int numOfPoints = 0;
        double centerx = 0;
        double centery = 0;
        double diff = 0.0;
        PointVector newCenter = null;
        double [] sums = new double[center.size()];
        for (PointVector p : points) {
            distance += measure.compute(center.points(), p.points());
            if (distance < min_dist) {
                min_dist = distance;
                newCenter = p;
            }
            numOfPoints++;
            sums = MathArrays.ebeAdd(p.points(), sums);
        }
        for (int i = 0; i < sums.length; i++) {
            sums[i] = sums[i] / numOfPoints;
        }
        System.out.println("Old center " + center + " new center: " + newCenter);
        context.write(new Text(newCenter.toString()) , new Text(new PointVector(sums).toString()));
    }
}