Waiting for answer This question has not been answered yet. You can hire a professional tutor to get the answer.
The current program only run one iteration of the KMeans algorithm. Please revise it (in the main function) to implement iterative processing, paste your code here, and briefly describe how it works.
The current program only run one iteration of the KMeans algorithm. Please revise it (in the main function) to implement iterative processing, paste your code here, and briefly describe how it works. Were you able to successfully compile and run your program (yes/no)?
import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.FileUtil;import org.apache.hadoop.fs.FSDataInputStream;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class KMeans { public static class KMMapper extends Mapper{ private double [][] _centroids; private IntWritable cid = new IntWritable(); public void setup(Mapper.Context context){ Configuration conf = context.getConfiguration(); String filename = conf.get("Centroids-file"); _centroids = loadCentroids(filename, conf); } public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { double [] vec = parseVector(value.toString()); cid.set(closest(vec)); context.write(cid, value); } private int closest(double [] v){ double mindist = dist(v, _centroids[0]); int label =0; for (int i=1; it){ mindist = t; label = i; } } return label; } } public static class KMReducer extends Reducer { // write output: cid \t centroid_vector private Text result = new Text(); public void reduce(IntWritable key, Iterable vectors, Context context ) throws IOException, InterruptedException { double [] sum = null; int n=0; for (Text vec : vectors) { double [] v = parseVector(vec.toString()); if (sum == null) sum = v; else for (int i = 0; i < v.length; i++) sum[i] += v[i]; n ++; } String out = Double.toString(sum[0]/n); for (int i = 1; i < sum.length; i ++ ){ out += "," + Double.toString(sum[i]/n); // csv output } result.set(out); context.write(key, result); } } // compute square Euclidean distance between two vectors v1 and v2 public static double dist(double [] v1, double [] v2){ double sum=0; for (int i=0; i< v1.length; i++){ double d = v1[i]-v2[i]; sum += d*d; } return Math.sqrt(sum); } // check convergence condition // max{dist(c1[i], c2[i]), i=1..numClusters < threshold private boolean converge(double [][] c1, double [][] c2, double threshold){ // c1 and c2 are two sets of centroids double maxv = 0; for (int i=0; i< c1.length; i++){ double d= dist(c1[i], c2[i]); if (maxv