package clusterassign;

import java.io.IOException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;

import datapoint.DataPoint;

/**
  * Mapper class for the Clluster Assign step
  */
public class ClusterAssignMapper extends Mapper<LongWritable, Text, DataPoint, DataPoint>
{
	/**
	  * ArrayList holding the k-Means Centroids read from a file
	  */
	public static ArrayList<DataPoint> kCentroids;

	/**
	  * Overridden setup method of Mapper class
	  * Parameters:	Context context
	  * Returns:	Nothing
	  * 
	  * Reads the file containing k-Means Centroids, parses it and loads the Centroids into the ArrayList kCentroids
	  */
	@Override
	public void setup(Context context)
		throws IOException, InterruptedException
	{
		// Call setup of super class
		super.setup(context);

		// Allocate memory for kCentroids
		kCentroids = new ArrayList<DataPoint>();

		// Get the context's configuration
		Configuration configuration = context.getConfiguration();

		// Open the file for reading
		Path path = new Path(configuration.get("fs.default.name") + configuration.get("kCentroidsFile"));
		FileSystem filesystem = FileSystem.get(configuration);
		BufferedReader reader = new BufferedReader(new InputStreamReader(filesystem.open(path)));

		String line = reader.readLine();
		while(line != null)
		{
			DataPoint centroid =  new DataPoint(line);
			kCentroids.add(centroid);
			line = reader.readLine();
		}
	}

	/**
	  * Overridden map function of Mapper Class
	  * Parameters:	LongWritable key, an offset in the file
	  * 			Text value, Data Points in a string format
	  *				Context context
	  * Returns:	(key, value) pairs where
	  *				key is a Canopy Center associated with the current Data Point
	  *				value is the Data Point being considered
	  * 
	  * The function receives a (key, value) pair and nparses it.
	  * For each k-Means Centroid in kCentroids, it calculates the centroid with the minimum distance.
	  * It outputs the pair (K-Centroid with minimum distance, Data Point)
	  */
	@Override 
	public void map(LongWritable key, Text value, Context context)
		throws IOException, InterruptedException
	{
		// Parse the Data Point
		DataPoint dataPoint = new DataPoint(value.toString());

		// Set up variables to find the centroid with minimum distance
		double minDistance = Double.MAX_VALUE;
		double distance;
		int offset = -1;

		// Find centroid with minimum distance
		for(int i = 0; i < kCentroids.size(); i++)
		{
			distance = dataPoint.complexDistance(kCentroids.get(i));
			if(distance < minDistance)
			{
				minDistance = distance;
				offset = i;
			}
		}

		// Write the pair (Centroid, Data Point)
		context.write(kCentroids.get(offset), dataPoint);
	}
}

@@ -5,21 +5,14 @@
 age 
-clusterassign
+source
 ;%0A%0Ai
@@ -407,24 +407,30 @@
 %0A/**%0A  *
+ clase
  Mapper 
 class fo
@@ -425,42 +425,36 @@
 per 
-class for the Clluster
+para el paso
  As
-s
 ign
- step
+ar Cluster
 %0A  *
@@ -552,359 +552,59 @@
 %0A%7B%0A%09
-/**%0A%09  * ArrayList holding the k-Means Centroids read from a file%0A%09  */%0A%09public static ArrayList%3CDataPoint%3E kCentroids;%0A%0A%09/**%0A%09  * Overridden setup method of Mapper class%0A%09  * Parameters:%09Context context%0A%09  * Returns:%09Nothing%0A%09  * %0A%09  * Reads the file containing k-Means Centroids, parses it and loads the Centroids into the ArrayList kCentroids%0A%09  */
+%0A%09public static ArrayList%3CDataPoint%3E kCentroids;%0A%0A%09
 %0A%09@O
@@ -698,37 +698,8 @@
 %09%7B%0A%09
-%09// Call setup of super class
 %0A%09%09s
@@ -730,26 +730,27 @@
 // A
-llocate memory for
+signar memoria para
  kCe
@@ -806,43 +806,8 @@
 ;%0A%0A%09
-%09// Get the context's configuration
 %0A%09%09C
@@ -870,36 +870,8 @@
 %0A%0A%09%09
-// Open the file for reading
 %0A%09%09P
@@ -1295,593 +1295,8 @@
 %7D%0A%0A%09
-/**%0A%09  * Overridden map function of Mapper Class%0A%09  * Parameters:%09LongWritable key, an offset in the file%0A%09  * %09%09%09Text value, Data Points in a string format%0A%09  *%09%09%09%09Context context%0A%09  * Returns:%09(key, value) pairs where%0A%09  *%09%09%09%09key is a Canopy Center associated with the current Data Point%0A%09  *%09%09%09%09value is the Data Point being considered%0A%09  * %0A%09  * The function receives a (key, value) pair and nparses it.%0A%09  * For each k-Means Centroid in kCentroids, it calculates the centroid with the minimum distance.%0A%09  * It outputs the pair (K-Centroid with minimum distance, Data Point)%0A%09  */
 %0A%09@O
@@ -1420,31 +1420,8 @@
 %7B%0A%09%09
-// Parse the Data Point
 %0A%09%09D
@@ -1480,71 +1480,8 @@
 ;%0A%0A%09
-%09// Set up variables to find the centroid with minimum distance
 %0A%09%09d
@@ -1563,46 +1563,8 @@
 %0A%0A%09%09
-// Find centroid with minimum distance
 %0A%09%09f
@@ -1764,48 +1764,8 @@
 %0A%0A%09%09
-// Write the pair (Centroid, Data Point)
 %0A%09%09c

Update ClusterAssignMapper.java