title: AWS EMR File Already exists: Hadoop Job reading and writing to S3
body: <p>I have a Hadoop job running in EMR and i am passing the S3 Path as input and output to this Job.</p>
  
  <p>When i run locally everything is working fine.( As there is a single node)</p>
  
  <p>How ever when i run in EMR with 5 node cluster i am running into File Already exists IO Exception.</p>
  
  <p>The output path has a timestamp in it so the out put path doesn't exists in S3. </p>
  
  <pre><code>Error: java.io.IOException: File already exists:s3://&lt;mybucket_name&gt;/8_9_0a4574ca-96d0-47c8-8eb8-4deb82944d4b/customer/RawFile12.txt/1523583593585/TOKENIZED/part-m-00000
  </code></pre>
  
  <p>I have a very simple hadoop app (primarily my mapper) which reads each line from a file and converts it (using an existing library) </p>
  
  <p>Not sure why each node is trying to write with the same file name.</p>
  
  <p>Here is mapper</p>
  
  <pre><code>public static class TokenizeMapper extends Mapper&lt;Object,Text,Text,Text&gt;{
          public void map(Object key, Text value,Mapper.Context context) throws IOException,InterruptedException{
              //TODO: Invoke Core Engine to transform the Data
              Encryption encryption = new Encryption();
              String tokenizedVal = encryption.apply(value.toString());
              context.write(tokenizedVal,1);
          }
      }
  </code></pre>
  
  <p>Any my Reducer</p>
  
  <pre><code>public static class TokenizeReducer extends Reducer&lt;Text,Text,Text,Text&gt; {
          public void reduce(Text text,Iterable&lt;Text&gt; lines,Context context) throws IOException,InterruptedException{
              Iterator&lt;Text&gt; iterator = lines.iterator();
              int counter =0;
              while(iterator.hasNext()){
                  counter++;
              }
  
              Text output = new Text(""+counter);
              context.write(text,output);
          }
      }
  </code></pre>
  
  <p>And my main class</p>
  
  <pre><code>public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
          long startTime = System.currentTimeMillis();
          try {
              Configuration config = new Configuration();
              String[] additionalArgs = new GenericOptionsParser(config, args).getRemainingArgs();
  
              if (additionalArgs.length != 2) {
                  System.err.println("Usage: Tokenizer Input_File and Output_File ");
                  System.exit(2);
              }
  
  
              Job job = Job.getInstance(config, "Raw File Tokenizer");
              job.setJarByClass(Tokenizer.class);
              job.setMapperClass(TokenizeMapper.class);
              job.setReducerClass(TokenizeReducer.class);
  
              job.setNumReduceTasks(0);
              job.setOutputKeyClass(Text.class);
              job.setOutputKeyClass(Text.class);
  
              FileInputFormat.addInputPath(job, new Path(additionalArgs[0]));
              FileOutputFormat.setOutputPath(job, new Path(additionalArgs[1]));
  
              boolean status = job.waitForCompletion(true);
              if (status) {
                  //System.exit(0);
                  System.out.println("Completed Job Successfully");
              } else {
                  System.out.println("Job did not Succeed");
              }
          }
          catch(Exception e){
              e.printStackTrace();
          }
          finally{
              System.out.println("Total Time for processing =["+(System.currentTimeMillis()-startTime)+"]");
          }
      }
  </code></pre>
  
  <p>I am passing the arguments when i am launching the cluster as</p>
  
  <pre><code>s3://&lt;mybucket&gt;/8_9_0a4574ca-96d0-47c8-8eb8-4deb82944d4b/customer/RawFile12.txt
  
  s3://&lt;mybucket&gt;/8_9_0a4574ca-96d0-47c8-8eb8-4deb82944d4b/customer/RawFile12.txt/1523583593585/TOKENIZED
  </code></pre>
  
  <p>Appreciate any inputs.</p>
  
  <p>Thanks</p>