title: Hadoop Mapreduce - Access Local File System
body: <p>I have a requirement where in the Map Reduce code should read the local file system in each node. The program will be running on HDFS and I cannot change the FileSystem property for hadoop in xml files for configuration.</p>
  
  <p>I have tried the following solutions, but none gave me results.</p>
  
  <p>Approach 1</p>
  
  <pre><code>Configuration config = new Configuration(); 
  FileSystem localFileSystem = FileSystem.get(config);
  localFileSystem.set("fs.defaultFS", "file:///");
  BufferedReader bufferRedaer = new BufferedReader(new InputStreamReader(localFileSystem.open(new Path("/user/input/localFile"))));
  </code></pre>
  
  <p>Approach 2</p>
  
  <pre><code>Configuration config = new Configuration();
  LocalFileSystem localFileSystem = FileSystem.getLocal(config);
  BufferedReader bufferRedaer = new BufferedReader(new InputStreamReader(localFileSystem.open(new Path("/user/input/localFile"))));
  </code></pre>
  
  <p>Approach 3</p>
  
  <pre><code>Configuration config = new Configuration(); 
  LocalFileSystem localFileSystem = FileSystem.getLocal(config);
  localFileSystem.set("fs.defaultFS", "file:///");
  BufferedReader bufferRedaer = new BufferedReader(new InputStreamReader(localFileSystem.open(new Path("/user/input/localFile"))));
  </code></pre>
  
  <p>Approach 4</p>
  
  <pre><code>Configuration config = new Configuration(); 
  LocalFileSystem localFileSystem = FileSystem.getLocal(config);
  BufferedReader bufferRedaer = new BufferedReader(new InputStreamReader(localFileSystem.getRaw().open(new Path("/user/input/localFile"))));
  </code></pre>
  
  <p>This did not work either
  [<a href="https://stackoverflow.com/questions/17072543/reading-hdfs-and-local-files-in-java][1]">Reading HDFS and local files in Java</a></p>
  
  <p>Each of them gave the error: No such file exists</p>
  
  <p>Error Stack</p>
  
  <pre><code>attempt_201406050021_0018_m_000000_2: java.io.FileNotFoundException: File /home/cloudera/sftp/id_rsa does not exist
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:468)
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:380)
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:231)
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:183)
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.LocalFileSystem.copyFromLocalFile(LocalFileSystem.java:81)
  attempt_201406050021_0018_m_000000_2:   at org.apache.hadoop.fs.FileSystem.copyFromLocalFile(FileSystem.java:1934)
  attempt_201406050021_0018_m_000000_2:   at com.skanda.ecomm.sftp.FTPMapper.configure(FTPMapper.java:91)
  </code></pre>
  
  <p>I am hoping to get a positive solution here. Let me know where I am going wrong.</p>
  
  <p>Main class (Driver class)</p>
  
  <pre><code>/*
   * @SFTPClient.java @May 20, 2014
   *
   * 
   */
  package com.skanda.ecomm.sftp;
  
  import java.net.URI;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.filecache.DistributedCache;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileOutputFormat;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  /**
   * 
   * &lt;p&gt;
   * SFTPClient Class
   * &lt;/p&gt;
   *
   * @author skanda
   * @version 1.0
   *
   */
  public class SFTPClient extends Configured implements Tool {
  
      public int run(String[] args) throws Exception {
  
          Configuration config = getConf();
  
          String inputPath = config.get(ApplicationConstants.INPUT_PATH);
          String outputPath = config.get(ApplicationConstants.OUTPUT_PATH);
          String configPath = config.get(ApplicationConstants.CONFIG_PATH);
          int reducers = Integer.parseInt(config.get(ApplicationConstants.REDUCERS));
  
          if(outputPath == null || inputPath == null || configPath == null) {
              throw new Exception("Usage: \n" + "-D configPath=&lt;configPath&gt; -D inputPath=&lt;inputPath&gt; -D reducers=&lt;reducers" +  
                      "-D outputPath=&lt;path&gt;");
          }
  
          JobConf conf = new JobConf(SFTPClient.class);
          conf.setJobName("SFTP Injection client");
  
          DistributedCache.addCacheFile(new URI(configPath),conf); 
  
          conf.setMapperClass(FTPMapper.class);
          conf.setReducerClass(FTPReducer.class);
  
          conf.setMapOutputKeyClass(IntWritable.class);
          conf.setMapOutputValueClass(Text.class);
  
          conf.setOutputKeyClass(IntWritable.class);
          conf.setOutputValueClass(IntWritable.class);
  
          // configuration should contain reference to your namenode
          FileSystem fs = FileSystem.get(new Configuration());
          fs.delete(new Path(outputPath), true); // true stands for recursively, deleting the folder you gave
  
          conf.setStrings(ApplicationConstants.INPUT_PATH, inputPath);
          conf.setStrings(ApplicationConstants.OUTPUT_PATH, outputPath);
  
          FileInputFormat.setInputPaths(conf, new Path(inputPath));       
          FileOutputFormat.setOutputPath(conf, new Path(outputPath));     
  
          conf.setNumReduceTasks(reducers);
          conf.setInt(ApplicationConstants.NUNBER_OF_REDUCERS, reducers);
  
          JobClient.runJob(conf);
  
          return 0;
      }        
  
      public static void main(String[] args) throws Exception {
          int exitCode = ToolRunner.run(new SFTPClient(), args);
          System.exit(exitCode);
      }
  }
  </code></pre>
  
  <p>Mapper</p>
  
  <pre><code>/*
   * @FTPMapper.java  @May 20, 2014
   *
   * 
   */
  package com.skanda.ecomm.sftp;
  
  import java.io.BufferedReader;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.net.InetAddress;
  import java.net.URI;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.filecache.DistributedCache;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.LocalFileSystem;
  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.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.MapReduceBase;
  import org.apache.hadoop.mapred.Mapper;
  import org.apache.hadoop.mapred.OutputCollector;
  import org.apache.hadoop.mapred.Reporter;
  
  import com.ftp.mapreduce.CommonUtility;
  import com.ftp.mapreduce.RetrieveFileNames;
  import com.jcraft.jsch.hm.Channel;
  
  /**
   * 
   * &lt;p&gt;
   * FTP Mapper Class
   * &lt;/p&gt;
   *
   * @author skanda
   * @version 1.0
   *
   */
  
  @SuppressWarnings("unused")
  public class FTPMapper extends MapReduceBase implements Mapper&lt;LongWritable, Text, IntWritable, Text&gt; {
  
      private URI[] localFiles;       
      private String userName;
      private String hostName;
      private String folderPath;
      private int reducers;
      private byte[] pvtKey;
      private String fileName;
      private String startDate;
      private String endDate;
      private String sshKeyPath;
      private String password;
  
      public void configure(JobConf job) {
          Properties properties = new Properties();
  
          try {
              localFiles = DistributedCache.getCacheFiles(job);
  
              if (localFiles != null &amp;&amp; localFiles.length == 1) {
                  Configuration conf = new Configuration();
  
  
  
                  FileSystem fileSystem = FileSystem.get(localFiles[0], conf);
                  BufferedReader bufferRedaer=new BufferedReader(new InputStreamReader(fileSystem.open(new Path(localFiles[0]))));
                  properties.load(bufferRedaer);
  
                  userName = properties.getProperty(ApplicationConstants.USER_NAME);
                  reducers = job.getInt(ApplicationConstants.NUNBER_OF_REDUCERS, 30);
                  hostName = properties.getProperty(ApplicationConstants.SFTP_SERVER_HOST);
                  folderPath = properties.getProperty(ApplicationConstants.HOSTFILE_DIRECTORY_PATH);
                  fileName = properties.getProperty(ApplicationConstants.FILE_NAME_PATTERN);
                  startDate = properties.getProperty(ApplicationConstants.FILE_START_DATE);
                  endDate = properties.getProperty(ApplicationConstants.FILE_END_DATE);
                  sshKeyPath = properties.getProperty(ApplicationConstants.SSH_KEY_PATH);
                  password = properties.getProperty(ApplicationConstants.PASSWORD);
  
                  System.out.println("--------------------------------------------------");
                  /*FileSystem fs = FileSystem.getLocal(conf);
                  //Path inputPath = fs.makeQualified(new Path(sshKeyPath));
                  String inputPath = new Path("file:///home/cloudera/"+sshKeyPath).toUri().getPath();
                  fs.copyFromLocalFile(new Path(inputPath), new Path("outputSFTP/idFile") );*/
                  try{
                      Configuration conf1 = new Configuration();
                      Path pt = new Path("file:///home/cloudera/.ssh/id_rsa");
                      FileSystem fs = FileSystem.get( new URI("file:///home/cloudera/.ssh/id_rsa"), conf);
                      LocalFileSystem localFileSystem = fs.getLocal(conf1);
                      BufferedReader bufferRedaer1 = new BufferedReader(new InputStreamReader(localFileSystem.open(pt)));
  
                      String str = null;
                      while ((str = bufferRedaer1.readLine())!= null)
                      {
                          System.out.println("-----------");
                          System.out.println(str);
                      }
                  }catch(Exception e){
                      System.out.println("failed again");
                      String computername=InetAddress.getLocalHost().getHostName();
                      System.out.println(computername);
                      e.printStackTrace();
                  }
  
                  System.out.println("--------------------------------------------------");
  
  
  
                  Configuration config = new Configuration(); 
                  config.set("fs.defaultFS", "file:////");
                  LocalFileSystem localFileSystem = FileSystem.getLocal(config);
                  bufferRedaer = new BufferedReader(new InputStreamReader(localFileSystem.open(new Path(sshKeyPath))));
  
                  /*Configuration config = new Configuration();
                  //config.set("fs.defaultFS", "file:///home/cloudera/.ssh/id_rsa");
                  LocalFileSystem fileSystm = FileSystem.getLocal(config);
                  Path path = fileSystm.makeQualified(new Path("/home/cloudera/.ssh/id_rsa"));*/
  
                  //FileInputFormat.setInputPaths(job, path);
  
                  //bufferRedaer = new BufferedReader(new InputStreamReader(fileSystem.open(path)));
  
                  String key = "";
  
                  try {
                        String line = "";
                        while ((line = bufferRedaer.readLine()) != null) {
                            key += line + "\n";   
                          }
                        pvtKey = key.getBytes();
                  } catch(Exception e){
                      e.printStackTrace();
                  } finally {
                      //fileSystem.close();
                      //bufferRedaer.close();
                  }       
              }
  
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
  
      public void map(LongWritable key, Text value, OutputCollector&lt;IntWritable, Text&gt; output, Reporter reporter)
          throws IOException {
  
          List&lt;String&gt; filterFileNamesList = new ArrayList&lt;String&gt;();
  
          Channel channel = CommonUtility.connectSFTP(userName, hostName, pvtKey);
  
          Map&lt;String, String&gt; fileNamesMap = CommonUtility.getFileNames(channel, folderPath);
  
          List&lt;String&gt; filterFileNameList_output = RetrieveFileNames.FILTER_BY_NAME.retrieveFileNames(fileNamesMap, filterFileNamesList, 
                                                              fileName, startDate, endDate);
  
          for (int i = 0; i &lt; filterFileNameList_output.size(); i++) {
              int keyGroup = i % reducers;
              output.collect(new IntWritable(keyGroup), new Text(filterFileNameList_output.get(i)));
          }
      }
  }
  </code></pre>