一、Hadoop压缩简介
11、hadoop的3个阶段 2 （1）分布式文件系统HDFS 3 （2）分布式编程框架MapReduce 4 （3）yarn框架 5 62、Hadoop数据压缩 7 MR操作过程中进行大量数据传输。 8 压缩技术能够有效的减少底层存储（HDFS）读写字节数。 9 压缩提高了网络带宽和磁盘空间的效率。 10 数据压缩能够有效的节省资源！ 11 压缩是mr程序的优化策略！ 12 通过压缩编码对mapper或者reducer数据传输进行数据的压缩，以减少磁盘IO。 13 143、压缩的基本原则 15 1、运算密集型任务少用压缩 16 2、IO密集型的任务，多用压缩 17 184、MR支持的压缩编码 19 压缩格式 | hadoop是否自带？ |文件拓展名 | 是否可以切分 20 DEFAULT | 是 | .deflate | 否 21 Gzip | 是 | .gz | 否 22 bzip2 | 是 | .bz2 | 是 23 LZO | 否 | .lzo | 是 24 Snappy | 否 | .snappy | 否 25 265、编码/解码器 27 DEFAULT | org.apache.hadoop.io.compress.DefaultCodeC 28 Gzip | org.apache.hadoop.io.compress.GzipCodeC 29 bzip2 | org.apache.hadoop.io.compress.BZip2CodeC 30 LZO | com.hadoop.compression.lzo.LzoCodeC 31 Snappy | org.apache.hadoop.io.compress.SnappyCodeC 32 336、压缩性能 34 压缩算法 | 原始文件大小 | 压缩文件大小| 压缩速度 | 解压速度 35 gzip | 8.3GB | 1.8GB |17,5MB/s |58MB/s 36 bzip2 | 8.3GB | 1.1GB |2.4MB/s |9.5MB/s 37 LZO | 8.3gb | 2.9GB |49.3MB/s |74.6MB/s 38 397、使用方式 40 （1）map端输出压缩 41 //开启map端的输出压缩 42 conf.setBoolean("mapreduce.map.output.compress", true); 43 //设置压缩方式 44 //conf.setClass("mapreduce.map.output.compress.codec", DefaultCodec.class, CompressionCodec.class); 45 conf.setClass("mapreduce.map.output.compress.codec",BZip2Codec.class, CompressionCodec.class); 46 （2）reduce端输出压缩 47 //开启reduce端的输出压缩 48 FileOutputFormat.setCompressOutput(job, true); 49 //设置压缩方式 50 //FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class); 51 //FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 52 FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
二、Hadoop压缩使用方式
1.Mapper类
1package com.css.compress; 2 3import java.io.IOException; 4 5import org.apache.hadoop.io.IntWritable; 6import org.apache.hadoop.io.LongWritable; 7import org.apache.hadoop.io.Text; 8import org.apache.hadoop.mapreduce.Mapper; 9 10public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ 11 12 // key 起始偏移量 value 数据 context 上下文 13 @Override 14 protected void map(LongWritable key, Text value, Context context) 15 throws IOException, InterruptedException { 16 // 1.读取数据 17 String line = value.toString(); 18 // 2.切割 hello hunter 19 String[] words = line.split(" "); 20 // 3.循环的写到下一个阶段<hello,1><hunter,1> 21 for (String w : words) { 22 context.write(new Text(w), new IntWritable(1)); 23 } 24 } 25}
2.Reducer类
1package com.css.compress; 2 3import java.io.IOException; 4 5import org.apache.hadoop.io.IntWritable; 6import org.apache.hadoop.io.Text; 7import org.apache.hadoop.mapreduce.Reducer; 8 9public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ 10 11 @Override 12 protected void reduce(Text key, Iterable<IntWritable> values, 13 Context context) throws IOException, InterruptedException { 14 // 1.统计单词出现的次数 15 int sum = 0; 16 // 2.累加求和 17 for (IntWritable count : values) { 18 // 拿到值累加 19 sum += count.get(); 20 } 21 // 3.结果输出 22 context.write(key, new IntWritable(sum)); 23 } 24}
3.Driver类
1package com.css.compress; 2 3import java.io.IOException; 4 5import org.apache.hadoop.conf.Configuration; 6import org.apache.hadoop.fs.Path; 7import org.apache.hadoop.io.IntWritable; 8import org.apache.hadoop.io.Text; 9import org.apache.hadoop.io.compress.BZip2Codec; 10import org.apache.hadoop.io.compress.CompressionCodec; 11import org.apache.hadoop.io.compress.DefaultCodec; 12import org.apache.hadoop.io.compress.GzipCodec; 13import org.apache.hadoop.mapreduce.Job; 14import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 15import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 16 17public class WordCountDriver { 18 public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 19 // 1.获取job信息 20 Configuration conf = new Configuration(); 21 Job job = Job.getInstance(conf); 22 23 // 开启map端的输出压缩 24 // conf.setBoolean("mapreduce.map.output.compress", true); 25 // 设置压缩方式 26 // conf.setClass("mapreduce.map.output.compress.codec", DefaultCodec.class, CompressionCodec.class); 27 // conf.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class); 28 29 // 2.获取jar包 30 job.setJarByClass(WordCountDriver.class); 31 // 3.获取自定义的mapper与reducer类 32 job.setMapperClass(WordCountMapper.class); 33 job.setReducerClass(WordCountReducer.class); 34 // 4.设置map输出的数据类型 35 job.setMapOutputKeyClass(Text.class); 36 job.setMapOutputValueClass(IntWritable.class); 37 // 5.设置reduce输出的数据类型（最终的数据类型） 38 job.setOutputKeyClass(Text.class); 39 job.setOutputValueClass(IntWritable.class); 40 41 // 开启reduce端的输出压缩 42 FileOutputFormat.setCompressOutput(job, true); 43 // 设置压缩方式 44 // FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class); 45 // FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 46 FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); 47 48 // 6.设置输入存在的路径与处理后的结果路径 49 FileInputFormat.setInputPaths(job, new Path("c:/compress1031/in")); 50 FileOutputFormat.setOutputPath(job, new Path("c:/compress1031/out2")); 51 // 7.提交任务 52 boolean rs = job.waitForCompletion(true); 53 System.out.println(rs?0:1); 54 } 55}
4.输入文件words.txt
1I love Beijing 2I love China 3Beijing is the capital of China
5.输出文件的名字分别如下
1（1） 2part-r-00000.bz2 3 4（2） 5part-r-00000.deflate 6 7（3） 8part-r-00000.gz
三、自定义压缩工具
1.自定义压缩工具类
1package com.css.compress; 2 3import java.io.File; 4import java.io.FileInputStream; 5import java.io.FileOutputStream; 6import java.io.IOException; 7 8import org.apache.hadoop.conf.Configuration; 9import org.apache.hadoop.io.IOUtils; 10import org.apache.hadoop.io.compress.CompressionCodec; 11import org.apache.hadoop.io.compress.CompressionOutputStream; 12import org.apache.hadoop.util.ReflectionUtils; 13 14public class TestCompress { 15 public static void main(String[] args) throws ClassNotFoundException, IOException { 16 compress("c:/compress1031/intest/test.txt","org.apache.hadoop.io.compress.DefaultCodec"); 17 compress("c:/compress1031/intest/test.txt","org.apache.hadoop.io.compress.BZip2Codec"); 18 compress("c:/compress1031/intest/test.txt","org.apache.hadoop.io.compress.GzipCodec"); 19 } 20 21 // 测试压缩方法 22 private static void compress(String fileName, String method) throws ClassNotFoundException, IOException{ 23 // 1.获取输入流 24 FileInputStream fis = new FileInputStream(new File(fileName)); 25 Class<?> cName = Class.forName(method); 26 CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(cName, new Configuration()); 27 // 2.输出流 28 FileOutputStream fos = new FileOutputStream(new File(fileName + codec.getDefaultExtension())); 29 // 3.创建压缩输出流 30 CompressionOutputStream cos = codec.createOutputStream(fos); 31 // 4.流的对拷 32 IOUtils.copyBytes(fis, cos, 1024*1024*2, false); 33 // 5.关闭资源 34 fis.close(); 35 cos.close(); 36 fos.close(); 37 } 38}
2.输入文件名
test.txt
3.输出文件名
1（1） 2test.txt.deflate 3 4（2） 5test.txt.bz2 6 7（3） 8test.txt.gz