大数据框架之Hadoop：MapReduce（四）Hadoop数据压缩

news2024/9/27 9:21:11

4.1概述

1、压缩概述

压缩技术能够有效减少底层存储系统（HDFS）读写字节数。压缩提高了网络带宽和磁盘空间的效率。在运行MR程序时，IO操作、网络数据传输、shuffle和Merge要花大量的时间，尤其是数据规模很大和工作负载密集的情况下，因此，使用数据压缩显得非常重要。

鉴于磁盘IO和网络带宽是Hadoop的宝贵资源，数据压缩对于节省资源、最小化磁盘IO和网络传输非常有帮助。可以在任意MapReduce阶段启用压缩。不过，尽管压缩和解压缩的CPU开销不高，其性能的提升和资源的节省并非没有代价。

2、压缩策略与原则

压缩是提高Hadoop运行效率的一种优化策略。

通过对Mapper、Reducer运行过程的数据进行压缩，以减少磁盘IO，提高MR程序运行速度。

👉 注意：采用压缩技术减少了磁盘IO，但同时增加了CPU运算负载。所以，压缩特性运用得当能提高性能，但运用不当也可能降低性能。

压缩基本原则：

  • 运算密集型的job，少用压缩
  • IO密集型的job，多用压缩

4.2MR支持的压缩编码

压缩格式hadoop自带？算法文件扩展名是否可切分换成压缩格式后，原来的程序是否需要修改
DEFLATE是，直接使用DEFLATE.deflate否和文本处理一样，不需要修改
Gzip是，直接使用DEFLATE.gz否和文本处理一样，不需要修改
bzip2是，直接使用bzip2.bz2是和文本处理一样，不需要修改
LZO否，需要安装LZO.lzo是需要建索引，还需要制定输入格式
Snappy否，需要安装Snappy.snappy否和文本处理一样，不需要修改

为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器，如下表所示。

压缩格式对应的编码/解码器
DEFLATEorg.apache.hadoop.io.compress.DefaultCodec
gziporg.apache.hadoop.io.compress.GzipCodec
bzip2org.apache.hadoop.io.compress.BZip2Codec
LZOcom.hadoop.compression.lzo.LzopCodec
Snappyorg.apache.hadoop.io.compress.SnappyCodec

压缩性能比较

压缩算法原始文件大小压缩文件大小压缩速度解压速度
gzip8.3GB1.8GB17.5MB/s58MB/s
bzip28.3GB1.1GB2.4MB/s9.5MB/s
LZO8.3GB2.9GB49.3MB/s74.6MB/s

4.3压缩方式选择

4.3.1Gzip压缩

优点：压缩率比较高，而且压缩解压缩速度也比较快；Hadoop本身支持，在应用中处理Gzip格式的文件和直接处理文本一样；大部分Linux系统都自带Gzip命令，使用方便。

缺点：不支持split。

应用场景：当每个文件压缩之后在130M以内时（一个块大小内），都可以考虑Gzip压缩格式。例如说一天或者一个小时的日志压缩成一个Gzip文件。

4.3.2Bzip2压缩

优点：支持split；具有很高的压缩率，比Gzip压缩率都高；Hadoop本身自带，使用方便。

缺点：压缩解压缩速度慢。

使用场景：适合对速度要求不高，但需要较高的压缩率的时候；或者输出之后的数据比较大，处理之后的数据需要压缩存档减少磁盘空间并且以后数据用得比较少的情况；或者对个单个很大的文本文件向要缩减少存储空间，同时又需要支持split，而且兼容之前的应用程序的情况。

4.3.3Lzo压缩

优点：压缩解压缩速度也比较快，合理的压缩率；支持split，是Hadoop中最流行的压缩方式；可以在Linux系统下安装lzop命令，使用方便。

缺点：压缩率比Gzip要低一些；Hadoop本身不支持，需要安装；在应用中对Lzo格式的文件需要做一些特殊处理（为了支持split需要建索引，还需要制定InputFormat的Lzo格式）。

应用场景：一个很大的文本文件，压缩之后还大于200M以上的可以考虑，而且单个文件越大，Lzo优点越明显。

4.3.4Snappy压缩

优点：高速压缩速度和合理的压缩率。

缺点：不支持split；压缩率比Gzip要低；Hadoop本身不支持，需要安装。

应用场景：当MapReduce作业的Map输出的数据比较大的时候，作为Map到Reduce的中间数据的压缩格式；或者作为一个MapReduce作业的输出和另一个MapReduce作业的输入。

4.4压缩位置选择

压缩可以在MapReduce作用的任意阶段启用，如下图所示。

Untitled

4.5压缩参数配置

参数默认值阶段建议
io.compression.codecs （在core-site.xml中配置）org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec输入压缩Hadoop使用文件扩展名判断是否支持某种编解码器
mapreduce.map.output.compress（在mapred-site.xml中配置）falsemapper输出这个参数设为true启用压缩
mapreduce.map.output.compress.codec（在mapred-site.xml中配置）org.apache.hadoop.io.compress.DefaultCodecmapper输出企业多使用LZO或Snappy编解码器在此阶段压缩数据
mapreduce.output.fileoutputformat.compress（在mapred-site.xml中配置）falsereducer输出这个参数设为true启用压缩
mapreduce.output.fileoutputformat.compress.codec（在mapred-site.xml中配置）org.apache.hadoop.io.compress.DefaultCodecreducer输出使用标准工具或者编解码器，如gzip和bzip2
mapreduce.output.fileoutputformat.compress.type（在mapred-site.xml中配置）RECORDreducer输出SequenceFile输出使用的压缩类型：NONE和BLOCK

4.6压缩实操案例

4.6.1数据流的压缩和解压缩

CompressionCodec有两个方法可以用于轻松地压缩或解压缩数据。

要想对正在被写入一个输出流的数据进行压缩，我们可以使用createOutputStream(OutputStreamout)方法创建一个CompressionOutputStream，将其以压缩格式写入底层的流。

相反，要想对输入流读取而来的数据进行解压缩，则调用createInputStream(InputStreamin)函数，从而获得一个CompressionInputSream，从而从底层的流读取未压缩的数据。

测试一下如下压缩方式：

名字编码
gziporg.apache.hadoop.io.compress.GzipCodec
bzip2org.apache.hadoop.io.compress.BZip2Codec
DEFLATEorg.apache.hadoop.io.compress.DefaultCodec
package com.cuiyf41.compress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;

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）关闭资源
        cos.close();
        fos.close();
        fis.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();
    }
}

4.6.2Map输出端采用压缩

即使你的MapReduce的输入输出文件都是未压缩的文件，你仍然可以对Map任务的中间结果输出做压缩，因为它要写在硬盘并且通过网络传输到Reduce节点，对其压缩可以提高很多性能，这些工作只要设置两个属性即可，我们来看下代码怎么设置。

1、给大家提供的Hadoop源码支持的压缩格式有：BZip2Codec 、DefaultCodec

package com.atguigu.mapreduce.compress;
import java.io.IOException;
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.BZip2Codec;	
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.output.FileOutputFormat;

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

2、Mapper保持不变

package com.cuiyf41.compress;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    Text k = new Text();
    IntWritable v = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // 1 获取一行
        String line = value.toString();

        // 2 切割
        String[] words = line.split(" ");

        // 3 循环写出
        for(String word:words){
            k.set(word);
            context.write(k, v);
    }
}

3、Reducer保持不变

package com.cuiyf41.compress;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    IntWritable v = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        int sum = 0;

        // 1 汇总
        for(IntWritable value:values){
            sum += value.get();
        }

        v.set(sum);

        // 2 输出
        context.write(key, v);
    }
}

4.6.3Reduce输出端采用压缩

基于WordCount案例处理。

1、修改驱动

package com.cuiyf41.compress;

import java.io.IOException;
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.BZip2Codec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.Lz4Codec;
import org.apache.hadoop.io.compress.SnappyCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

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

2、Mapper和Reducer保持不变（详见4.6.2）

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/382112.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

3月2日第壹简报，星期四，农历二月十一

3月2日第壹简报，星期四，农历二月十一

3月2日第壹简报&#xff0c;星期四&#xff0c;农历二月十一坚持阅读&#xff0c;静待花开1. 第三次延期&#xff01;财政部&#xff1a;对于购置期在2023年的新能源汽车继续免征车辆购置税。2. 我国2月份制造业PMI为52.6% &#xff0c;创2012年4月以来新高。3. 全国地级市财力…
阅读更多...
深入浅出消息队列MSMQ

深入浅出消息队列MSMQ

消息队列MSMQ&#xff0c;相信稍有开发经验的小伙伴都了解一些。开始讲解之前&#xff0c;我们先弄清楚一件事&#xff0c;为什么我们要使用MSMQ&#xff1a; 您可能认为您能够通过一个简单的数据库表(一个应用程序往其中写入数据&#xff0c;另一个应用程序从中读取数据)来应用…
阅读更多...
采编式AIGC视频生产流程编排实践

采编式AIGC视频生产流程编排实践

作者 | 百度人工智能创作团队 导读 本文从业务出发&#xff0c;系统介绍了采编式 TTV的实现逻辑和实现路径。结合业务拆解&#xff0c;实现了一个轻量级服务编排引擎&#xff0c;有效实现业务诉求、高效支持业务扩展。 全文6451字&#xff0c;预计阅读时间17分钟。 01 背景 近…
阅读更多...
WebRTC之RTP封装与解封装

WebRTC之RTP封装与解封装

1 前言rtp_rtcp模块作为Webrtc组件中非常重要的组成部分&#xff0c;首先是对应rtp和rtcp的封装与解封装&#xff0c;第二部分是对QOS各种策略的支持都离不开rtcp的控制协议。这里首先进行协议封装的探讨。2 RTP协议解析各个音视频的大佬对下面这张RTP协议图应该并不陌生&#…
阅读更多...
ChatGPT到底是个啥？简析ChatGPT！

ChatGPT到底是个啥？简析ChatGPT！

目录 ​编辑 1. ChatGPT到底是个啥&#xff1f; 1.1. 简介 1.2 玩法 1.2.1.生成公司理念、生成广告标语 1.2.2.写小说写故事写情书 1.2.3.生成自媒体文案 1.2.4.写代码 2.简析ChatGPT 2.1.ChatGPT核心能力 2.2.ChatGPT进化史 2.2.1.历史沿革 2.2.2.算法 2.3.ChatGPT特…
阅读更多...
k8s学习之路 | Pod 基础

k8s学习之路 | Pod 基础

文章目录Pod 基础认知什么是 PodPod 的示例 yamlPod 的形式Pod 的多容器协同Pod 的生命周期容器的类型应用容器初始化容器临时容器静态 Pod什么是静态 Pod静态 Pod 位置Pod 探针机制探针类型Probe 配置项探针案例Pod 基础认知 什么是 Pod https://kubernetes.io/zh-cn/docs/c…
阅读更多...
Word处理控件Aspose.Words功能演示：使用 Java 合并 MS Word 文档

Word处理控件Aspose.Words功能演示：使用 Java 合并 MS Word 文档

Aspose.Words 是一种高级Word文档处理API&#xff0c;用于执行各种文档管理和操作任务。API支持生成&#xff0c;修改&#xff0c;转换&#xff0c;呈现和打印文档&#xff0c;而无需在跨平台应用程序中直接使用Microsoft Word。此外&#xff0c; Aspose API支持流行文件格式处…
阅读更多...
Overlay网络技术

Overlay网络技术

大家好&#xff0c;我是技福的小咖老师。 Overlay网络是通过网络虚拟化技术&#xff0c;在同一张Underlay网络上构建出的一张或者多张虚拟的逻辑网络。不同的Overlay网络虽然共享Underlay网络中的设备和线路&#xff0c;但是Overlay网络中的业务与Underlay网络中的物理组网和互…
阅读更多...
aardio - 【库】简单信息框

aardio - 【库】简单信息框

昨晚得知aardio作者一鹤的妻子病情严重&#xff0c;深感悲痛。今日给一鹤捐赠少许&#xff0c;望其妻能挺过难关&#xff0c;早日康复。 aardio是一个很好的编程工具&#xff0c;我非常喜欢&#xff0c;这两年也一直在用。虽然未曾用其获利&#xff0c;但其灵活的语法&#xff…
阅读更多...
操作系统真相还原——第7章 中断

操作系统真相还原——第7章 中断

中断&#xff1a;CPU 暂停正在执行的程序&#xff0c;转而去执行处理该事件的程序&#xff0c;当这段程序执行完毕后&#xff0c; CPU 继续执行刚才的程序。 通常&#xff0c;中断牺牲的是个体的时间&#xff0c;但可以实现多设备的并发&#xff0c;从而提高系统效率 操作系统…
阅读更多...
评估Jupyter环境的安全性

评估Jupyter环境的安全性

评估Jupyter环境的安全性 如何判断您的 Jupyter 实例是否安全&#xff1f; NVIDIA AI 红队开发了一个 JupyterLab 扩展来自动评估 Jupyter 环境的安全性。 jupysec 是一种根据近 100 条规则评估用户环境的工具&#xff0c;这些规则检测配置和工件&#xff0c;这些配置和工件已被…
阅读更多...
暴力递归到动态规划

暴力递归到动态规划

暴力递归到动态规划 假设有排成一行的n个位置&#xff0c; 记为1~n&#xff0c;n-定大于或等于2。开始时机器人在其中的m位置上(m 一定是1~n中的一个)。如果机器人来到1位置&#xff0c;那么下一步只能往右来到2位置&#xff1b;如果机器人来到n位置&#xff0c; 那么下一步只能…
阅读更多...
js中splice方法和slice方法

js中splice方法和slice方法

splice方法用来操作数组splice(startIndex,deleteNum,item1,....,)此操作会改变原数组。删除数组中元素参数解释&#xff1a;startIndex为起始index索引。deleteNum为从startIndex索引位置开始需要删除的个数。分三种情况&#xff1a;没有传第三个参数的情况下&#xff0c;dele…
阅读更多...
pytest两种生成测试报告的方法——html

pytest两种生成测试报告的方法——html

pytest有两种生成测试报告的方法&#xff08;html和allure&#xff09;&#xff0c;今天就给大家一介绍下html 一.pytest-html基本语法 1.安装&#xff1a;pip install pytest-html 2.查看版本&#xff1a;pip show pytest-html 3.生成测试报告基本语法&#xff1a; 语法一…
阅读更多...
STM32物联网项目之程序框架

STM32物联网项目之程序框架

前言&#xff1a; 这个系列&#xff0c;我主要写我用32f103实现的各种功能模块&#xff0c;已经程序编写过程中&#xff0c;硬件调试中出现的问题&#xff0c;一边记录&#xff0c;一边分享&#xff0c;一边复盘。 使用的是STM32cubemax&#xff0c;自动生成代码&#xff0c;…
阅读更多...
每日学术速递3.2

每日学术速递3.2

CV - 计算机视觉 | ML - 机器学习 | RL - 强化学习 | NLP 自然语言处理 Subjects: cs.CV 1.Interactive Segmentation as Gaussian Process Classification(CVPR 2023) 标题&#xff1a;作为高斯过程分类的交互式分割 作者&#xff1a;Minghao Zhou, Hong Wang, Qian Zha…
阅读更多...
tensorflow1.14.0安装教程--保姆级

tensorflow1.14.0安装教程--保姆级

//方法不止一种&#xff0c;下面仅展示一种。 注&#xff1a;本人电脑为win11&#xff0c;anaconda的python版本为3.9&#xff0c;但tensorflow需要python版本为3.7&#xff0c;所以下面主要阐述将python版本改为3.7后的安装过程以及常遇到的问题。 1.首先电脑安装好anaconda…
阅读更多...
java进阶—多线程

java进阶—多线程

学习线程&#xff0c;我们先来了解了解什么是进程&#xff1f;什么是线程 进程&#xff1a;就是在操作系统中运行的程序 线程&#xff1a;就是进程的一个执行单元&#xff0c;或者一条执行路劲 比如&#xff1a;我们打开应用商店&#xff0c;这个应用商店就是一个进程&#…
阅读更多...
字节实习二面

字节实习二面

网络体系结构分层&#xff08;7、5、4&#xff09; 答&#xff1a; OSI七层网络体系结构&#xff1a;物理层、数据链路层、网络层、传输层、会话层、表示层、应用层 TCP/IP四层网络体系结构&#xff1a;物理层、网际层、传输层、应用层 TCP/IP五层网络体系结构&#xff1a;物…
阅读更多...
ShopWind 多商户商城更新，Vue 3 前后端分离，页面自定义装修

ShopWind 多商户商城更新，Vue 3 前后端分离，页面自定义装修

本次为 V4 版本更新&#xff0c;新系统架构(技术栈)vue3 vite (打包编译工具) Composition API(组合式 API setup) Element Plus vueRouter (路由) 第三方组件&#xff1a;axios (数据请求) wangeditor(编辑器)&#xff0c;都是通过接口访问数据&#xff0c;页面效果更佳了…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023