title: Can't access hashmap in mapper, MapReduce
body: <p>I'd like to replace values of input data in my mapper, using dictionalies(csv) defined in another file. So I tried to put the csv data to HashMap and refer it in the mapper.</p>
  
  <p>The java code and csv below are simplified version of my program. This code <strong>works in my local environment</strong>(Mac OS X, pseudo-distributed mode), but <strong>doesn't in my EC2 instance(ubuntu, pseudo-distributed mode)</strong>.</p>
  
  <p>In detail, I got this stdout in process:</p>
  
  <pre><code>cat:4
  human:2
  flamingo:1
  </code></pre>
  
  <p>this means the filereader successfully put csv data into HashMap.</p>
  
  <p>However the mapper mapped nothing and therefore I got empty output in the EC2 environment, although it mapped 3 * (the number of lines of the input file) elements and generated the following in the local:</p>
  
  <pre><code>test,cat
  test,flamingo
  test,human
  </code></pre>
  
  <p>Does anyone have answers or hints? </p>
  
  <p><strong>Test.java</strong></p>
  
  <pre><code>import java.io.IOException;
  import java.util.StringTokenizer;
  import java.io.FileReader;
  import java.io.BufferedReader;
  import java.io.DataInput; 
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Map.Entry;
  
  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.io.Writable;
  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;
  import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.mapreduce.RecordWriter;
  import org.apache.hadoop.io.WritableUtils;
  
  public class Test {
  
    public static HashMap&lt;String, Integer&gt; map  = new HashMap&lt;String, Integer&gt;();
  
    public static class Mapper1 extends Mapper&lt;LongWritable, Text, Text, Text&gt; {
      @Override
      protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        for(Map.Entry&lt;String, Integer&gt; e : map.entrySet()) {
          context.write(new Text(e.getKey()), new Text("test"));
        }
      }
    }
  
    public static class Reducer1 extends Reducer&lt;Text, Text, Text, Text&gt; {
      @Override
      protected void reduce(Text key, Iterable&lt;Text&gt; vals, Context context) throws IOException, InterruptedException {
        context.write(new Text("test"), key);
      }
    }
  
    public static class CommaTextOutputFormat extends TextOutputFormat&lt;Text, Text&gt; {
      @Override
      public RecordWriter&lt;Text, Text&gt; getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        Configuration conf = job.getConfiguration();
        String extension = ".txt";
        Path file = getDefaultWorkFile(job, extension);
        FileSystem fs = file.getFileSystem(conf);
        FSDataOutputStream fileOut = fs.create(file, false);
        return new LineRecordWriter&lt;Text, Text&gt;(fileOut, ",");
      }
    }
  
    public static void get_list(String list_path){
      try {
        FileReader fr = new FileReader(list_path);
        BufferedReader br = new BufferedReader(fr);
        String line = null, name = null;
        int leg = 0;
  
        while ((line = br.readLine()) != null) {
          if (!line.startsWith("name") &amp;&amp; !line.trim().isEmpty()) {
            String[] name_leg = line.split(",", 0);
            name = name_leg[0];
            leg = Integer.parseInt(name_leg[1]);
            map.put(name, leg);
          }
        }
        br.close();
      }
      catch(IOException ex) {
        System.err.println(ex.getMessage());
        ex.printStackTrace();
      }
  
      for(Map.Entry&lt;String, Integer&gt; e : map.entrySet()) {
        System.out.println(e.getKey() + ":" + e.getValue());
      }
    }
  
    public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
  
      if (args.length != 3) {
        System.err.println(
          "Need 3 arguments: &lt;input dir&gt; &lt;output base dir&gt; &lt;list path&gt;");
        System.exit(1);
      }
  
      get_list(args[2]);
      Job job = Job.getInstance(conf, "test");
  
      job.setJarByClass(Test.class);
      job.setMapperClass(Mapper1.class);
      job.setReducerClass(Reducer1.class);
      job.setNumReduceTasks(1);
      job.setInputFormatClass(TextInputFormat.class);
  
      // mapper output
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(Text.class);
  
      // reducer output
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
  
      // formtter
      job.setOutputFormatClass(CommaTextOutputFormat.class);
  
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
      if(!job.waitForCompletion(true)){
        System.exit(1);
      }
  
      System.out.println("All Finished");
      System.exit(0);
    }
  }
  </code></pre>
  
  <p><strong>list.csv (args[2])</strong></p>
  
  <pre><code>name,legs
  cat,4
  human,2
  flamingo,1
  </code></pre>
  
  <h3>=================================</h3>
  
  <p>I refer to @Rahul Sharma 's answer and modifiy my code as below. Then my code works in the both environments. </p>
  
  <p>Thank you very much @Rahul Sharma and @Serhiy for your precise answer and useful comments.</p>
  
  <p><strong>Test.java</strong></p>
  
  <pre><code>import java.io.IOException;
  import java.util.StringTokenizer;
  import java.io.FileReader;
  import java.io.BufferedReader;
  import java.io.DataInput; 
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.net.URI;
  import java.io.InputStreamReader;
  
  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.io.Writable;
  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;
  import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.mapreduce.RecordWriter;
  import org.apache.hadoop.io.WritableUtils;
  
  public class Test {
  
    public static HashMap&lt;String, Integer&gt; map  = new HashMap&lt;String, Integer&gt;();
  
    public static class Mapper1 extends Mapper&lt;LongWritable, Text, Text, Text&gt; {
  
      @Override
      protected void setup(Context context) throws IOException, InterruptedException {
        URI[] files = context.getCacheFiles();
        Path list_path = new Path(files[0]);
  
        try {
          FileSystem fs = list_path.getFileSystem(context.getConfiguration());
          BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(list_path)));
          String line = null, name = null;
          int leg = 0;
  
          while ((line = br.readLine()) != null) {
            if (!line.startsWith("name") &amp;&amp; !line.trim().isEmpty()) {
              String[] name_leg = line.split(",", 0);
              name = name_leg[0];
              leg = Integer.parseInt(name_leg[1]);
              map.put(name, leg);
            }
          }
          br.close();
        }
        catch(IOException ex) {
          System.err.println(ex.getMessage());
          ex.printStackTrace();
        }
  
        for(Map.Entry&lt;String, Integer&gt; e : map.entrySet()) {
          System.out.println(e.getKey() + ":" + e.getValue());
        }
      }
  
      @Override
      protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        for(Map.Entry&lt;String, Integer&gt; e : map.entrySet()) {
          context.write(new Text(e.getKey()), new Text("test"));
        }
      }
  
    }
  
    public static class Reducer1 extends Reducer&lt;Text, Text, Text, Text&gt; {
      @Override
      protected void reduce(Text key, Iterable&lt;Text&gt; vals, Context context) throws IOException, InterruptedException {
        context.write(new Text("test"), key);
      }
    }
  
    // Writer
    public static class CommaTextOutputFormat extends TextOutputFormat&lt;Text, Text&gt; {
      @Override
      public RecordWriter&lt;Text, Text&gt; getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        Configuration conf = job.getConfiguration();
        String extension = ".txt";
        Path file = getDefaultWorkFile(job, extension);
        FileSystem fs = file.getFileSystem(conf);
        FSDataOutputStream fileOut = fs.create(file, false);
        return new LineRecordWriter&lt;Text, Text&gt;(fileOut, ",");
      }
    }
  
    public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
  
      if (args.length != 3) {
        System.err.println(
          "Need 3 arguments: &lt;input dir&gt; &lt;output base dir&gt; &lt;list path&gt;");
        System.exit(1);
      }
  
      Job job = Job.getInstance(conf, "test");
      job.addCacheFile(new Path(args[2]).toUri());
  
      job.setJarByClass(Test.class);
      job.setMapperClass(Mapper1.class);
      job.setReducerClass(Reducer1.class);
      job.setNumReduceTasks(1);
      job.setInputFormatClass(TextInputFormat.class);
  
      // mapper output
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(Text.class);
  
      // reducer output
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
  
      // formtter
      job.setOutputFormatClass(CommaTextOutputFormat.class);
  
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
      if(!job.waitForCompletion(true)){
        System.exit(1);
      }
  
      System.out.println("All Finished");
      System.exit(0);
    }
  }
  </code></pre>