簡潔而不簡單

Hadoop資料壓縮

資料壓縮優點和缺點

? 壓縮技術能夠有效減少底層存盤系統（HDFS）讀寫位元組數，壓縮提高了網路帶寬和磁盤空間的效率，在 Hadoop下，尤其是資料規模很大和作業負載密集的情況下，使用資料壓縮顯得非常重要，在這種情況下，IO操作和網路資料傳輸要花大量的時間，還有， Shuffle與 Merge程序同樣也面臨著巨大的IO壓力鳘于磁盤IO和網路帶寬是 Hadoop的寶貴資源，資料壓縮對于節省資源、最小化磁盤IO和網路傳輸非常有幫助，

? 不過，盡管壓縮與解壓操作的CPU開銷不髙，其性能的提升和資源的節省并非沒有代價，如果磁盤IO和網路帶寬影響了 MapReduce作業性能，在任意 MapReduce階段啟用壓縮都可以改善端到端處理時間并減少IO和網路流量，

壓縮策略和原則

? 壓縮是提高 Hadoop運行效率的一種優化策略通過對 Mapper、 Reducer運行程序的資料進行壓縮，以減少磁盤IO，提高MR程式運行速度，
? 注意：釆用壓縮技術減少了磁盤IO，但同時增加了CPU運算負擔，所以，壓縮特性運用得當能提高性能，但運用不當也可能降低性能壓碩訓本原則：

（1）運算密集型的job，少用壓縮
（2）IO密集型的job，多用壓縮！！

MR支持的壓縮編碼

為了支持多種壓縮/解壓縮演算法，Hadoop引入了編碼/解碼器，如下表所示，

壓縮性能的比較

壓縮方式選擇

Gzip壓縮

Bzip2壓縮

Lzo壓縮

Snappy壓縮

壓縮位置選擇

壓縮引數配置

壓縮案例

public class TestCompress {

	public static void main(String[] args) throws Exception {
		compress("e:/hello.txt","org.apache.hadoop.io.compress.BZip2Codec");
//		decompress("e:/hello.txt.bz2");
	}

	// 1、壓縮
	private static void compress(String filename, String method) throws Exception {
		
		// （1）獲取輸入流
		FileInputStream fis = new FileInputStream(new File(filename));
		
		Class codecClass = Class.forName(method);
		
		CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, new Configuration());
		
		// （2）獲取輸出流
		FileOutputStream fos = new FileOutputStream(new File(filename +codec.getDefaultExtension()));
		CompressionOutputStream cos = codec.createOutputStream(fos);
		
		// （3）流的對拷
		IOUtils.copyBytes(fis, cos, 1024*1024*5, false);
		
		// （4）關閉資源
		fis.close();
		cos.close();
		fos.close();
	}

	// 2、解壓縮
	private static void decompress(String filename) throws FileNotFoundException, IOException {
		
		// （0）校驗是否能解壓縮
		CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());

		CompressionCodec codec = factory.getCodec(new Path(filename));
		
		if (codec == null) {
			System.out.println("cannot find codec for file " + filename);
			return;
		}
		
		// （1）獲取輸入流
		CompressionInputStream cis = codec.createInputStream(new FileInputStream(new File(filename)));
		
		// （2）獲取輸出流
		FileOutputStream fos = new FileOutputStream(new File(filename + ".decoded"));
		
		// （3）流的對拷
		IOUtils.copyBytes(cis, fos, 1024*1024*5, false);
		
		// （4）關閉資源
		cis.close();
		fos.close();
	}
}

Map輸出端采用壓縮

public class WordCountDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		Configuration configuration = new Configuration();

		// 開啟map端輸出壓縮
	configuration.setBoolean("mapreduce.map.output.compress", true);
		// 設定map端輸出壓縮方式
	configuration.setClass("mapreduce.map.output.compress.codec", BZip2Codec.class, CompressionCodec.class);

		Job job = Job.getInstance(configuration);

		job.setJarByClass(WordCountDriver.class);

		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		boolean result = job.waitForCompletion(true);

		System.exit(result ? 1 : 0);
	}
}

Mapper和Reducer代碼不變

Reduce輸出端采用壓縮

public class WordCountDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		
		Configuration configuration = new Configuration();
		
		Job job = Job.getInstance(configuration);
		
		job.setJarByClass(WordCountDriver.class);
		
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		// 設定reduce端輸出壓縮開啟
		FileOutputFormat.setCompressOutput(job, true);
		
		// 設定壓縮的方式
	    FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class); 
//	    FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); 
//	    FileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class); 
	    
		boolean result = job.waitForCompletion(true);
		
		System.exit(result?1:0);
	}
}

相關資料

本文配套GitHub：https://github.com/zhutiansama/FocusBigData

標籤：Java

上一篇：tp5 paginate() 動態分頁隱藏欄位

下一篇：JAVA Math類