java - 使用 hadoop mapreduce 识别文件夹中的文件扩展名以处理文件

Question

我需要处理一个包含文本文件的文件夹。文本文件可以是任何扩展名。

对于每个扩展，我们需要单独的自定义读取器来在 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 文件，则可以正常工作。

我是否遗漏了什么？

希望这不会正常工作，因为我正在遍历文件夹(比如文件夹 2--> Data5.pdf、Data4.csv)。这个 setInputFormatClass 不会对TextInputFormat.addInputPath(job, in) 作为输入格式将设置为 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);

我正在努力实现与上面类似的目标。但这没有给出任何输出。请提出建议。

Answer 1

从 Context 中获取 Input Split，然后从路径中获取 Path & name。

Context => getInputSplit() => getPath => getName()

获得名称后，从该索引中找到 lastIndexOf(".") 和子字符串。

现在您在子字符串中有了扩展名，并用它来进行比较。

编辑:

以下方法对您可行吗？

1. 为每种类型的扩展设置单独的 Mapper。

2. 在 Driver 类中添加以下行。

   MultipleInputs.addInputPath(job, path_pdf,inputFormatClass, PDFMapper.class)
   
   MultipleInputs.addInputPath(job, path_xml, inputFormatClass,XMLMapper.class)
   
   MultipleInputs.addInputPath(job, path_html,inputFormatClass,HTMLMapper.class)
   
   MultipleInputs.addInputPath(job, path_csv,inputFormatClass,CVSMapper.class)