我需要处理一个文本文件文件夹。文本文件可以是任何扩展名。
对于每个扩展，我们需要单独的自定义读取器，以便在hadoop中处理该文件。

folder1/
   Data1.pdf
   Data2.xml
   Data3.html
   Data4.txt
   Data5.csv

在文件夹中获取文件扩展名并为mr作业设置自定义输入格式的更好方法是什么？
到目前为止我所做的是
司机

FileStatus[] stati = null;
try {
    stati = fs.listStatus(in);
} catch (FileNotFoundException e) {
    e.printStackTrace();
} catch (IOException e) {
    e.printStackTrace();
}
for (FileStatus status : stati) {
    Path path = status.getPath();
    System.out.println("Path----> "+path);
    /*
     * get file extension
     */
    String ext = FilenameUtils.getExtension(path.toString());
    System.out.println("ext--->"+ext);
    if(ext.equals("pdf")){
          //custom pdf record reader
         job.setInputFormatClass(PdfInputFormat.class);
    }
    else{
         job.setInputFormatClass(TextInputFormat.class);
    }
}

但是这在folder1中不起作用，但是如果folder1只包含.pdf文件，这就可以了。
我错过什么了吗？
希望这不会正常工作，因为我正在遍历文件夹（例如folder2-->data5.pdf，data4.csv）。此setinputformatclass不会对textinputformat显示任何影响。addinputpath（job，in）因为inputformat将使用data4.csv的上一次迭代值设置。

job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(NullWritable.class);
    job.setMapperClass(MyMapper.class);
    job.setReducerClass(MyReducer.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    try {
        TextInputFormat.addInputPath(job, in);
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    TextOutputFormat.setOutputPath(job, out);

编辑

Job job = null;
try {
 job = new Job(conf, "TextMining");
} catch (IOException e) {
 e.printStackTrace();
}
/*
 * check entension
 */
for (FileStatus status : stati) {
 Path path = status.getPath();
 System.out.println("Path----> "+path);
 /*
  * get file extension
  */
 String ext = FilenameUtils.getExtension(path.toString());
 System.out.println("ext--->"+ext);
 if(ext.equals("pdf")){
       System.out.println("Pdf File Format");
      // MultipleInputs.addInputPath(job, path,PdfInputFormat.class, PDFStemmingMapper.class);
       job.setInputFormatClass(PdfInputFormat.class);
 }
 else if(ext.equals("xlsx")){ 
       System.out.println("Excel File Format");
       job.setInputFormatClass(ExcelInputFormat.class);
 }
 else{
       System.out.println("normal Text File");
       job.setInputFormatClass(TextInputFormat.class);
 }
}
job.setJarByClass(Driver.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(NullWritable.class);
//job.setInputFormatClass(TextInputFormat.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
job.setOutputFormatClass(TextOutputFormat.class);

//   try {
  //    TextInputFormat.addInputPath(job, in);
  //   } catch (IOException e) {
  //    e.printStackTrace();
  //   }
    TextOutputFormat.setOutputPath(job, out);

我正在努力实现上面的目标。但这并没有产生任何效果。请建议。