title: How to use multiple CSV files in mapreduce
body: <p>First, I will explain what I am trying to do. First, I am putting input file (first CSV file) into mapreduce job and other CSV file will be put inside mapper class. But here is the thing. The code in mapper class does not work properly, like this right bottom code. I want to combine two CSV files to use several columns in each CSV file.
  For example, 1 file has BibNum (user account), checkoutdatetime (book checkoutdatetime), and itemtype (book itemtype), and 2 CSV file has BibNum (user account), Title (book Title), Itemtype and so on. I want to find out which book will be likely borrowed in coming month. I would be really appreciated if you know the way can combine two CSV file and enlighten me with any helps. If you have any doubts for my code, just let me know, I will try to clarify it.</p>
  
  <pre><code> Path p = new Path("hdfs://0.0.0.0:8020/user/training/Inventory_Sample");
         FileSystem fs = FileSystem.get(conf);
  
         BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(p)));
  
  
          try {
  
  
              String BibNum = "Test";
              //System.out.print("test");
              while(br.readLine() != null){
                  //System.out.print("test");
                  if(!br.readLine().startsWith("BibNumber")) {
                      String subject[] = br.readLine().split(",");
                      BibNum = subject[0];
                  }
              }
  </code></pre>
  
  <p>.</p>
  
  <pre><code>import java.io.BufferedReader;
  
  import java.io.FileReader;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.util.Date;
  import java.util.HashMap;
  import java.text.ParseException;
  import java.text.SimpleDateFormat;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FileSystem;
  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 StubMapper extends Mapper&lt;LongWritable, Text, Text, IntWritable&gt; {
  
      private Text outkey = new Text();
      //private MinMaxCountTuple outTuple = new MinMaxCountTuple();
  
      //String csvFile = "hdfs://user/training/Inventory_Sample";
      @Override
      public void map(LongWritable key, Text value, Context context)
              throws IOException, InterruptedException {
  
  
         Configuration conf = context.getConfiguration();
         //conf.addResource("/etc/hadoop/conf/core-site.xml");
         //conf.addResource("/etc/hadoop/conf/hdfs-site.xml");
         Path p = new Path("hdfs://0.0.0.0:8020/user/training/Inventory_Sample");
         FileSystem fs = FileSystem.get(conf);
  
         BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(p)));
  
  
          try {
  
  
              String BibNum = "Test";
              //System.out.print("test");
              while(br.readLine() != null){
                  //System.out.print("test");
                  if(!br.readLine().startsWith("BibNumber")) {
                      String subject[] = br.readLine().split(",");
                      BibNum = subject[0];
                  }
              }
  
              if(value.toString().startsWith("BibNumber"))
              {
                  return;
              }
  
              String data[] = value.toString().split(",");
  
              String BookType = data[2];
              String DateTime = data[5];
  
              SimpleDateFormat frmt = new SimpleDateFormat("MM/dd/yyyy hh:mm:ss a");
  
              Date creationDate = frmt.parse(DateTime);
              frmt.applyPattern("dd-MM-yyyy");
              String dateTime = frmt.format(creationDate);
  
              //outkey.set(BookType + " " + dateTime);
              outkey.set(BibNum + " " + BookType + " " + dateTime);
  
              //outUserId.set(userId);
              context.write(outkey, new IntWritable(1));
  
          } catch (Exception e) {
              // TODO Auto-generated catch block
              e.printStackTrace();
          }
          finally{
              br.close();
          }
  
  
      }
  }
  </code></pre>