KMeansWithCombiner.java

import java.io.IOException;
import java.util.*;
import java.io.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.GenericOptionsParser;

public class KMeansWithCombiner {

    public static class KMeansMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
        private IntWritable one_key = new IntWritable();
        private int n_centers;
        private double[] center_x;
        private double[] center_y;

        protected void setup(Context context) throws IOException, InterruptedException {
            Configuration conf = context.getConfiguration();
            n_centers = conf.getInt("n_centers", -1);
            center_x = new double[n_centers];
            center_y = new double[n_centers];

            for (int i = 0; i < n_centers; i++) {
                center_x[i] = conf.getFloat("center_x_" + i, 0);
                center_y[i] = conf.getFloat("center_y_" + i, 0);
            }
        }

        // calculate distance between two points..
        public double getDist(double x1, double y1, double x2, double y2) {
            double dist = (x1 - x2) * (x1 - x2) + (y1 - y2) * (y1 - y2);
            return Math.sqrt(dist);
        }

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());

            if (itr.countTokens() < 2) return;
            if (n_centers == 0) return;

            double x = Double.parseDouble(itr.nextToken().trim());
            double y = Double.parseDouble(itr.nextToken().trim());
            int cluster_idx = 0;

            double min = Integer.MAX_VALUE;
            for (int i = 0; i < n_centers; i++) {
                if (getDist(center_x[i], center_y[i], x, y) <= min) {
                    min = getDist(center_x[i], center_y[i], x, y);
                    cluster_idx = i;
                }
            }
            one_key.set(cluster_idx);
            context.write(one_key, value);
        }
    }

    public static class KMeansCombiner extends Reducer<IntWritable, Text, IntWritable, Text> {

        public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            double x_total = 0;
            double y_total = 0;
            int cnt = 0;
            for (Text val : values) {
                StringTokenizer itr = new StringTokenizer(val.toString());
                x_total += Double.parseDouble(itr.nextToken().trim());
                y_total += Double.parseDouble(itr.nextToken().trim());
                cnt++;
            }
            context.write(key, new Text(x_total + " " + y_total + " " + cnt));
        }
    }

    public static class KMeansReducer extends Reducer<IntWritable, Text, IntWritable, Text> {

        public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            double x_total = 0;
            double y_total = 0;
            int cnt = 0;
            for (Text val : values) {
                StringTokenizer itr = new StringTokenizer(val.toString());
                x_total += Double.parseDouble(itr.nextToken().trim());
                y_total += Double.parseDouble(itr.nextToken().trim());
                cnt += Integer.parseInt(itr.nextToken().trim());
            }
            Text result = new Text((x_total / (double) cnt) + " " + (y_total / (double) cnt));
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        int n_centers = 2;
        int n_iter = 3;

        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: KMeansWithCombiner <in> <out>");
            System.exit(2);
        }
        initCenters(conf, n_centers);

        for (int i = 0; i < n_iter; i++) {
            Job job = new Job(conf, "KMeansWithCombiner");
            job.setJarByClass(KMeans.class);
            job.setMapperClass(KMeansMapper.class);
            job.setCombinerClass(KMeansCombiner.class); // Set the Combiner class
            job.setReducerClass(KMeansReducer.class);

            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(Text.class);

            job.setInputFormatClass(TextInputFormat.class);
            job.setOutputFormatClass(TextOutputFormat.class);

            FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
            FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            FileSystem.get(job.getConfiguration()).delete(new Path(otherArgs[1]), true);
            job.waitForCompletion(true);

            updateCenters(conf, n_centers); // Calculate the new centers
        }
    }

    public static void initCenters(Configuration conf, int n_centers) {
        conf.setInt("n_centers", n_centers);
        for (int i = 0; i < n_centers; i++) {
            conf.setFloat("center_x_" + i, (float) (1.0 / (double) n_centers));
            conf.setFloat("center_y_" + i, (float) (1.0 / (double) n_centers));
        }
    }

    public static void updateCenters(Configuration conf, int n_centers) throws Exception {
        FileSystem dfs = FileSystem.get(conf);
        Path filenamePath = new Path("/user/bigdata/output0611_3/part-r-00000");
        FSDataInputStream in = dfs.open(filenamePath);
        BufferedReader reader = new BufferedReader(new InputStreamReader(in));

        String line = reader.readLine();
        while (line != null) {
            StringTokenizer itr = new StringTokenizer(new String(line));
            int cluster_id = Integer.parseInt(itr.nextToken().trim());
            double x = Double.parseDouble(itr.nextToken().trim());
            double y = Double.parseDouble(itr.nextToken().trim());
            conf.setFloat("center_x_" + cluster_id, (float) x);  // Read and store the newly calculated center value
            conf.setFloat("center_y_" + cluster_id, (float) y);  // Read and store the newly calculated center value
            line = reader.readLine();
        }
    }
}