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

Question

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

对于每个扩展，我们需要单独的自定义阅读器来处理hadoop中的文件。

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

什么是更好的方式来获得文件夹中文件的扩展名并为我的MR Job设置Custom InputFormat？

我做了什么至今

驱动

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); 
    } 
} 

但这不会witin的folder1工作，但如果文件夹1只包含.pdf文件能正常工作。

我错过了什么吗？

希望，因为我在文件夹迭代这是行不通的罚款（比如文件夹2 - > Data5.pdf，Data4.csv）。这setInputFormatClass不会向 TextInputFormat.addInputPath任何影响（作业中，）作为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); 

我想实现有点像上面。 但是这没有给出任何输出。 请建议。

Answer 1

对于映射器，预期的输入是一条记录（由值表示）。 InputFormat是如何构造这个记录并传递给map方法的。

例：默认输入格式的TextInputFormat，将考虑一条线在一个文件中作为记录。这通常适用于txt/xsv文件。

对于其他类型的文件，更好的办法是使用自定义的InputFormat，它知道如何代表一个记录。 （在XML中，记录可以是一个子块）

如果你已经有InputFormat类你已经确定的所有文件类型，那么你可以使用MultipleInputs。

看看这里的Javadoc https://hadoop.apache.org/docs/r2.7.0/api/org/apache/hadoop/mapreduce/lib/input/MultipleInputs.html

识别扩展名可以用FileSystem API可

Answer 2

从上下文中来完成，获得输入分流，然后从路径获取路径&名。

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

一旦你的名字，找到该索引lastIndexOf(".")和子。

现在，您在子字符串中有扩展名并将其用于比较。

编辑：

是下面的方法对你是否可行？

1. 对于每种类型的扩展都有单独的映射器。

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)