title: Problems with setting up and accessing Distributed Cache
body: <p>For some reason I can't find any good sources online for getting Distributed Cache working with the new API. Hoping someone here can explain what I'm doing wrong. My current attempt is sort of a mish-mash of various things I've found online.</p>
  
  <p>This program attempts to run the k-nearest neighbors algorithm. The input file is the test dataset, while the distributed cache holds the train dataset and train labels. The mapper should take one row of test data, compare it to every row in the distributed cache data, and return the label of the row it is most similar to.</p>
  
  <pre><code>import java.net.URI;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.filecache.DistributedCache;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  
  public class KNNDriver extends Configured implements Tool {
      public int run(String[] args) throws Exception {
          if (args.length != 2) {
              System.out.printf("Usage: %s [generic options] &lt;input dir&gt; &lt;output dir&gt;\n", getClass().getSimpleName());
              return -1;
          }
  
          Configuration conf = new Configuration();
          // conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", "^");
  
          conf.setInt ("train_rows",1000);
          conf.setInt ("test_rows",1000);
          conf.setInt ("cols",612);
          DistributedCache.addCacheFile(new URI("cacheData/train_sample.csv"),conf);
          DistributedCache.addCacheFile(new URI("cacheData/train_labels.csv"),conf);
  
          Job job = new Job(conf);
          job.setJarByClass(KNNDriver.class); 
          job.setJobName("KNN");
  
          FileInputFormat.setInputPaths(job, new Path(args[0]));
          FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
          job.setMapperClass(KNNMapper.class);
          job.setReducerClass(KNNReducer.class);
          // job.setInputFormatClass(KeyValueTextInputFormat.class);
  
          job.setMapOutputKeyClass(IntWritable.class);
          job.setMapOutputValueClass(IntWritable.class);
  
          job.setOutputKeyClass(IntWritable.class);
          job.setOutputValueClass(IntWritable.class);
  
          boolean success = job.waitForCompletion(true);
          return success ? 0 : 1;
      }
  
      public static void main(String[] args) throws Exception {
          int exitCode = ToolRunner.run(new Configuration(), new KNNDriver(), args);
          System.exit(exitCode);
      }
  }
  
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.IOException;
  import java.io.FileNotFoundException;
  import java.util.Scanner;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Mapper;
  
  public class KNNMapper extends Mapper&lt;LongWritable, Text, IntWritable, IntWritable&gt; {
  
    int[][] train_vals;
    int[] train_label_vals;
    int train_rows;
    int test_rows;
    int cols;
  
    @Override
    public void setup(Context context) throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();
  
        // Path[] cacheFiles = context.getLocalCacheFiles();
  
        int train_rows = conf.getInt("train_rows", 0);
        int test_rows = conf.getInt("test_rows", 0);
        int cols = conf.getInt("cols", 0);
  
        train_vals = new int[train_rows][cols];
        train_label_vals = new int[train_rows];
  
        // read train csv, parse, and store into 2d int array
        Scanner myScan;
          try {
              myScan = new Scanner(new File("train_sample.csv"));
  
              //Set the delimiter used in file
              myScan.useDelimiter("[,\r\n]+");
  
              //Get all tokens and store them in some data structure
              //I am just printing them
  
              System.out.println("myScan loaded for train_sample");
  
              for(int row = 0; row &lt; train_rows; row++) {
                  for(int col = 0; col &lt; cols; col++) {
                      train_vals[row][col] = Integer.parseInt(myScan.next().toString());
  
                  }
              }
  
              myScan.close();
  
          } catch (FileNotFoundException e) {
              System.out.print("Error: Train file not found.");
          }
  
      // read train_labels csv, parse, and store into 2d int array
          try {
              myScan = new Scanner(new File("train_labels.csv"));
  
              //Set the delimiter used in file
              myScan.useDelimiter("[,\r\n]+");
  
              //Get all tokens and store them in some data structure
              //I am just printing them
  
              System.out.println("myScan loaded for train_sample");
  
  
              for(int row = 0; row &lt; train_rows; row++) {
                      train_label_vals[row] = Integer.parseInt(myScan.next().toString());
              }
  
              myScan.close();
  
          } catch (FileNotFoundException e) {
              System.out.print("Error: Train Labels file not found.");
          }
    }
  
    @Override
    public void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {
  
          // setup() gave us train_vals &amp; train_label_vals.
          // Each line in map() represents a test observation.  We iterate 
          // through every train_val row to find nearest L2 match, then
          // return a key/value pair of &lt;observation #, 
  
          // convert from Text to String
          String line = value.toString();
          long distance;
          double best_distance = Double.POSITIVE_INFINITY;
          int col_num;
  
          int best_digit = -1;
          IntWritable rowId = null;
          int i;
          IntWritable rowNum;
          String[] pixels;
  
          // comma delimited files, split on commas
          // first we find the # of rows
          for (i = 0; i &lt; train_rows; i++) {
              distance = 0;
              col_num = 0;
              pixels = line.split(",");
              rowId = new IntWritable(Integer.parseInt(pixels[0]));
  
              for (int j = 1; j &lt; cols; j++) {
                  distance += (Integer.parseInt(pixels[j]) - train_vals[i][j-1])^2;
              }
              if (distance &lt; best_distance) {
                  best_distance = distance;
                  best_digit = train_label_vals[i];
              }
          }
          context.write(rowId, new IntWritable(best_digit));
    }
  }
  </code></pre>
  
  <p>I commented out the Path... statement because I don't understand what it does, or how it sends the file data to the mapper, but I noticed it listed on a couple websites.  Currently the program is not finding the Distributed Cache datasets even though they are uploaded to HDFS.</p>