title: MapReduce canopy clustering centers
body: <p>I'm trying to understand this code for canopy clustering. The purpose of these two classes (one map, one reduce) is to find canopy centers. My problem is that I don't understand the difference between the map and reduce functions. They are nearly the same.</p>
  
  <p>So is there a difference? Or am I just repeating the same same process again in the reducer? </p>
  
  <p>I think the answer is that there is a difference in how the map and reduce functions handle the code. They perform different actions on the data even with similar code.</p>
  
  <p>So can someone please explain the process of the map and reduce when we try to find the canopy centers?</p>
  
  <p>I know for example that a map might looks like this -- (joe, 1) (dave, 1) (joe, 1) (joe, 1)</p>
  
  <p>and then the reduce will go like this: --- (joe, 3) (dave, 1)</p>
  
  <p>Does the same type of thing happen here?</p>
  
  <p>Or maybe I'm performing the same task twice? </p>
  
  <p>Thanks so much.</p>
  
  <p>map function:</p>
  
  <pre><code>package nasdaq.hadoop;
  
  import java.io.*;
  import java.util.*;
  
  import org.apache.hadoop.io.*;
  import org.apache.hadoop.mapreduce.*;
  import org.apache.hadoop.util.*;
  
  public class CanopyCentersMapper extends Mapper&lt;LongWritable, Text, Text, Text&gt; {
      //A list with the centers of the canopy
      private ArrayList&lt;ArrayList&lt;String&gt;&gt; canopyCenters;
  
  @Override
  public void setup(Context context) {
          this.canopyCenters = new ArrayList&lt;ArrayList&lt;String&gt;&gt;();
  }
  
  @Override
  public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
      //Seperate the stock name from the values to create a key of the stock and a list of values - what is list of values?
      //What exactly are we splitting here?
      ArrayList&lt;String&gt; stockData = new ArrayList&lt;String&gt;(Arrays.asList(value.toString().split(","))); 
  
      //remove stock and make first canopy center around it canopy center
      String stockKey = stockData.remove(0);
  
      //?
      String stockValue = StringUtils.join(",", stockData);
  
      //Check wether the stock is avaliable for usage as a new canopy center
      boolean isClose = false;    
  
      for (ArrayList&lt;String&gt; center : canopyCenters) {    //Run over the centers
  
      //I think...let's say at this point we have a few centers. Then we have our next point to check.
      //We have to compare that point with EVERY center already created. If the distance is larger than EVERY T1
      //then that point becomes a new center! But the more canopies we have there is a good chance it is within
      //the radius of one of the canopies...
  
              //Measure the distance between the center and the currently checked center
              if (ClusterJob.measureDistance(center, stockData) &lt;= ClusterJob.T1) {
                      //Center is too close
                      isClose = true;
                      break;
              }
      }
  
      //The center is not smaller than the small radius, add it to the canopy
      if (!isClose) {
          //Center is not too close, add the current data to the center
          canopyCenters.add(stockData);
  
          //Prepare hadoop data for output
          Text outputKey = new Text();
          Text outputValue = new Text();
  
          outputKey.set(stockKey);
          outputValue.set(stockValue);
  
          //Output the stock key and values to reducer
          context.write(outputKey, outputValue);
      }
  }
  </code></pre>
  
  <p>}</p>
  
  <p>Reduce function:</p>
  
  <pre><code>    package nasdaq.hadoop;
  
  import java.io.*;
  import java.util.*;
  
  import org.apache.hadoop.io.*;
  import org.apache.hadoop.mapreduce.*;
  
  public class CanopyCentersReducer extends Reducer&lt;Text, Text, Text, Text&gt; {
      //The canopy centers list
      private ArrayList&lt;ArrayList&lt;String&gt;&gt; canopyCenters;
  
  @Override
  public void setup(Context context) {
          //Create a new list for the canopy centers
          this.canopyCenters = new ArrayList&lt;ArrayList&lt;String&gt;&gt;();
  }
  
  public void reduce(Text key, Iterable&lt;Text&gt; values, Context context) throws IOException, InterruptedException {
      for (Text value : values) {
          //Format the value and key to fit the format
          String stockValue = value.toString();
          ArrayList&lt;String&gt; stockData = new ArrayList&lt;String&gt;(Arrays.asList(stockValue.split(",")));
          String stockKey = key.toString();
  
          //Check wether the stock is avaliable for usage as a new canopy center
          boolean isClose = false;    
          for (ArrayList&lt;String&gt; center : canopyCenters) {    //Run over the centers
                  //Measure the distance between the center and the currently checked center
                  if (ClusterJob.measureDistance(center, stockData) &lt;= ClusterJob.T1) {
                          //Center is too close
                          isClose = true;
                          break;
                  }
          }
  
          //The center is not smaller than the small radius, add it to the canopy
          if (!isClose) {
              //Center is not too close, add the current data to the center
              canopyCenters.add(stockData);
  
              //Prepare hadoop data for output
              Text outputKey = new Text();
              Text outputValue = new Text();
  
              outputKey.set(stockKey);
              outputValue.set(stockValue);
  
              //Output the stock key and values to reducer
              context.write(outputKey, outputValue);
          }
  
  
      }
  </code></pre>
  
  <p><em>*</em>*Edit -- more code and explanation</p>
  
  <p>Stockkey is the key value representing stocks. (nasdaq and things like that)</p>
  
  <p>ClusterJob.measureDistance():</p>
  
  <pre><code>    public static double measureDistance(ArrayList&lt;String&gt; origin, ArrayList&lt;String&gt; destination)
  {
      double deltaSum = 0.0;
      //Run over all points in the origin vector and calculate the sum of the squared deltas
      for (int i = 0; i &lt; origin.size(); i++) {
          if (destination.size() &gt; i) //Only add to sum if there is a destination to compare to
          {
              deltaSum = deltaSum + Math.pow(Math.abs(Double.valueOf(origin.get(i)) - Double.valueOf(destination.get(i))),2);
          }
      }
      //Return the square root of the sum
      return Math.sqrt(deltaSum);
  </code></pre>