Hadoop Map-Reduce的压缩最终输出文件

实现

还是以以前做的删选最高气温的例子为参照：

以前的例子可以见这篇文章：http://www.linuxidc.com/Linux/2012-05/61196.htm

我们现在要求让结果输出为压缩格式，所以保持Map类(MaxTemperatureMapper）和Reduce类（MaxTemperatureReducer）不变，只要在Job类的Configuration作一些压缩的配置即可，见第45-49行所示：

 
   

  
    
package com.charles.parseweather.compression; 
  
    
 
  
    
 
  
    
import org.apache.hadoop.conf.Configuration; 
  
    
import org.apache.hadoop.fs.Path; 
  
    
import org.apache.hadoop.io.IntWritable; 
  
    
import org.apache.hadoop.io.Text; 
  
    
import org.apache.hadoop.io.compress.CompressionCodec; 
  
    
import org.apache.hadoop.io.compress.GzipCodec; 
  
    
import org.apache.hadoop.mapreduce.Job; 
  
    
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
  
    
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 
  
    
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
  
    
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 
  
    
 
  
    
 
  
    
/** 
  
    
 *  
  
    
 * 
  
    
 * Description: 这个类定义并且运行作业,压缩版 
  
    
 * 
  
    
 * @author charles.wang 
  
    
 * @created May 24, 2012 5:29:12 PM 
  
    
 * 
  
    
 */ 
  
    
 
  
    
public class MaxTemperatureWithCompression { 
  
    
 
  
    
    /** 
  
    
     * @param args 
  
    
     */ 
  
    
    public static void main(String[] args) throws Exception{ 
  
    
        // TODO Auto-generated method stub 
  
    
 
  
    
         
  
    
        if (args.length !=2){ 
  
    
            System.err.println("Usage: MaxTemperature <input path> <output path>"); 
  
    
            System.exit(-1); 
  
    
        } 
  
    
         
  
    
        //创建一个Map-Reduce的作业 
  
    
        Configuration conf = new Configuration(); 
  
    
        conf.set("hadoop.job.ugi", "hadoop-user,hadoop-user"); 
  
    
         
  
    
        //在这里我们配置一些和压缩有关的参数 
  
    
         
  
    
        //我们设定reduce输出结果使用gzip压缩的形式 
  
    
        conf.setBoolean("mapred.output.compress", true); 
  
    
        conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class); 
  
    
         
  
    
         
  
    
         
  
    
        Job job = new Job(conf,"Get Maximum Weather Information with Compression! ^_^"); 
  
    
         
  
    
       
  
    
         
  
    
         
  
    
         
  
    
        //设定作业的启动类/  
  
    
        job.setJarByClass(MaxTemperatureWithCompression.class); 
  
    
         
  
    
        //解析输入和输出参数，分别作为作业的输入和输出，都是文件 
  
    
        FileInputFormat.addInputPath(job, new Path(args[0])); 
  
    
        FileOutputFormat.setOutputPath(job, new Path(args[1])); 
  
    
        
  
    
        //配置作业，设定Mapper类，Reducer类 
  
    
        job.setMapperClass(MaxTemperatureMapper.class); 
  
    
        job.setReducerClass(MaxTemperatureReducer.class); 
  
    
        job.setOutputKeyClass(Text.class); 
  
    
       job.setOutputValueClass(IntWritable.class); 
  
    
        
  
    
        
  
    
      
  
    
         
  
    
        System.exit(job.waitForCompletion(true)?0:1); 
  
    
        
  
    
         
  
    
         
  
    
         
  
    
         
  
    
 
  
    
    } 
  
    
 
  
    
} 
 
   

要运行这个例子，我们需要给出输入文件，因为Hadoop系统可以根据输入文件的扩展名自动识别基本文件，所以我们创建目录结构，并且上传一个gzip格式的文件作为map-reduce过程的输入：

然后我们运行的main中传入HDFS的输入文件和输出目录：

当执行完成之后，我们就可以在HDFS文件系统中看到最终的输出结果了，正如我们所预期的，这个结果是个gzip格式的文件：