【Hadoop】利用MultipleOutputs,MultiOutputFormat实现以不同格式输出到多个文件

　　这是小D 第一篇博客，有什么错误还请各位指正。

　　

　　小D 也是刚接触Hadoop ，因为在淘宝实习，有很多算法要在分布式环境下实现，所以这几天一直在看Hadoop，边用边学。

最近实现的一个算法需要reduce输出很多参数，每个参数的格式不一样，而且要做为下一次mapreduce的输入，大家都知道Hadoop的分布式操作系统HFS是以目录为节点读取文件的，每个reduce输出一个分片，所以必须把要输出的文件根据类型的不同输出到不同的目录中去。因为淘宝还在用 Hadoop 0.19的API ，所以没办法直接用MultipleOutputs 的write方法实现。

write

public void write(KEYOUT key,
                  VALUEOUT value,
                  String baseOutputPath)
           throws IOException,
                  InterruptedException
Write key value to an output file name. Gets the record writer from job's output format. Job's output format should be a FileOutputFormat.
Parameters:
key - the key
value - the value
baseOutputPath - base-output path to write the record to. Note: Framework will generate unique filename for the baseOutputPath
Throws:
IOException
InterruptedException

MultipleOutputs 没有类似”String baseOutputPath“ 属性。在0.19的API中MultipleOutputs只可以为不同类型的文件加上不同的前缀。比如先在main 函数指定

MultipleOutputs.addNamedOutput(datafliter, "rating", TextOutputFormat.class, Text.class, Text.class);

MultipleOutputs.addNamedOutput(datafliter, "bu", TextOutputFormat.class, Text.class, Text.class);

然后在reduce里实例化MultipleOutputs

private MultipleOutputs mos ;

...

public void configure(JobConf conf) {
        mos = new MultipleOutputs(conf);
...
}
public void close() throws IOException {  
        mos.close();  
 } 


public void reduce(Text key, Iterator<Text> values,
            OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
....

  OutputCollector<Text, Text> rateoutput =mos.getCollector("rating", reporter);
        rateoutput.collect(new Text(key), new Text(sb.toString()));

　　OutputCollector<Text, Text> rateoutput =mos.getCollector("bu", reporter);
        rateoutput.collect(new Text(key), new Text(sb.toString()));

}

这样可以实现多个输出，在文件前加上相应的前缀。

但是这样做很多不同格式的文件还是会在一个output目录下，苦思冥想了半天，查各种资料，问周围同事，终于发现MapRedeuce 框架了还有一个货是专门干这个的 那就是MultipleOutputFormat ， 它有两个子类MultipleTextOutputFormat 和MultipleSequenceOutputFormat 对应两种输出格式。但是想用这货还挺麻烦，要写一个子类继承他，重写父类的方法。

以我写的MultiOutputFormatByFileName 为例。这个类还可以定制key ，value 键值对的规则，这里我们不讨论这个，我们用到的是这两个方法

protected String generateLeafFileName(String name)

protected String generateFileNameForKeyValue(Text key, Text value, String name) 

第一个是根据原文件名生成一个新文件名，注意这里的新文件名可以包含目录，也就是可以写成 XX/XX/XX.dat,这样就可以生成新的目录。

第二个是根据文件中的键值对来生成文件名。也可以生成目录。

package org.taobao.edp.tcif.titleExtract.util;

import java.io.File;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;

public class MultiOutputFormatByFileName extends MultipleTextOutputFormat<Text, Text> {
    
    
    
    @Override
    protected String generateLeafFileName(String name) {
        // TODO Auto-generated method stub
        System.out.println(name);
        String[] names = name.split("-");
        
        return names[0]+File.separator+name;
    }
    
    @Override
    protected String generateFileNameForKeyValue(Text key, Text value,
            String name) {
        // TODO Auto-generated method stub
        return super.generateFileNameForKeyValue(key, value, name);
    }
    
}

OK 我们重写第一个方法，之前我们通过MultipleOutput 生成了带前缀的文件名，现在我们把这个前缀提取出来作为目录名，代码如上。

然后我们修改 main

MultipleOutputs.addNamedOutput(init,"q", MultiOutputFormatByFileName.class , Text.class, Text.class);
MultipleOutputs.addNamedOutput(init,"x", MultiOutputFormatByFileName.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(init,"bi", MultiOutputFormatByFileName.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(init,"bu", MultiOutputFormatByFileName.class, Text.class, Text.class);     

用自己的MultiOutputFormatByFileName 替换框架的TextOutputFormat

这样就可以实现多目录输出喽！

真心希望公司的分布式平台能支持Hadoop 0.21的API ， 一个简单的需求折腾了半天。