reducer.java

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class KMeansReducer extends Reducer<IntWritable, PointWritable, NullWritable, PointWritable> {

    private List<Point> means;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // Initialize the means from the distributed cache or any other source
        // Here, we assume the means are already set and available in the `means` list
        means = new ArrayList<>(); // Initialize and populate the means list
    }

    @Override
    protected void reduce(IntWritable key, Iterable<PointWritable> values, Context context)
            throws IOException, InterruptedException {
        // Accumulate the points associated with the key (closest mean index)
        List<Point> accumulatedPoints = new ArrayList<>();
        for (PointWritable value : values) {
            accumulatedPoints.add(value.getPoint().clone());
        }

        // Recompute the mean for this key
        Point newMean = computeMean(accumulatedPoints);

        // Update the means list with the new mean
        means.set(key.get(), newMean);

        // Emit the new mean as the output
        context.write(NullWritable.get(), new PointWritable(newMean));
    }

    private Point computeMean(List<Point> points) {
        // Compute the centroid of the points
        // Here, you can implement the centroid computation logic based on your Point class

        // Return the computed mean
    }
}