你好,游客 登录 注册 搜索
背景:
阅读新闻

hadoop中使用lzo的压缩

[日期:2014-07-21] 来源:www.xingxingge.com/  作者:心情 [字体: ]

  在hadoop中使用lzo的压缩算法可以减小数据的大小和数据的磁盘读写时间,不仅如此,lzo是基于block分块的,这样他就允许数据被分解成chunk,并行的被hadoop处理。这样的特点,就可以让lzo在hadoop上成为一种非常好用的压缩格式。

  lzo本身不是splitable的,所以当数据为text格式时,用lzo压缩出来的数据当做job的输入是一个文件作为一个map。但是 sequencefile本身是分块的,所以sequencefile格式的文件,再配上lzo的压缩格式,就可实现lzo文件方式的 splitable。

  由于压缩的数据通常只有原始数据的1/4,在HDFS中存储压缩数据,可以使集群能保存更多的数据,延长集群的使用寿命。不仅如此,由于 mapreduce作业通常瓶颈都在IO上,存储压缩数据就意味这更少的IO操作,job运行更加的高效。但是,在hadoop上使用压缩也有两个比较麻烦的地方:第一,有些压缩格式不能被分块,并行的处理,比如gzip。第二,另外的一些压缩格式虽然支持分块处理,但是解压的过程非常的缓慢,使job的瓶颈转移到了cpu上,例如bzip2。比如我们有一个1.1GB的gzip文件,该文件 被分成128MB/chunk存储在hdfs上,那么它就会被分成9块。为了能够在mapreduce中并行的处理各个chunk,那么各个mapper之间就有了依赖。而第二个mapper就会在文件的某个随机的byte出进行处理。那么gzip解压时要用到的上下文字典就会为空,这就意味这gzip的压缩文件无法在hadoop上进行正确的并行处理。也就因此在hadoop上大的gzip压缩文件只能被一个mapper来单个的处理,这样就很不高效,跟不用mapreduce没有什么区别了。而另一种bzip2压缩格式,虽然bzip2的压缩非常的快,并且甚至可以被分块,但是其解压过程非常非常的缓慢,并且不能被用streaming来读取,这样也无法在hadoop中高效的使用这种压缩。即使使用,由于其解压的低效,也会使得job的瓶颈转移到 cpu上去。

  如果能够拥有一种压缩算法,即能够被分块,并行的处理,速度也非常的快,那就非常的理想。这种方式就是lzo。lzo的压缩文件是由许多的小的 blocks组成(约256K),使的hadoop的job可以根据block的划分来splitjob。不仅如此,lzo在设计时就考虑到了效率问题,它的解压速度是gzip的两倍,这就让它能够节省很多的磁盘读写,它的压缩比的不如gzip,大约压缩出来的文件比gzip压缩的大一半,但是这样仍然比没有经过压缩的文件要节省20%-50%的存储空间,这样就可以在效率上大大的提高job执行的速度。以下是一组压缩对比数据,使用一个8.0GB的未经过压缩的数据来进行对比:

 

  可以看出,lzo压缩文件会比gzip压缩文件稍微大一些,但是仍然比原始文件要小很多倍,并且lzo文件压缩的速度几乎相当于gzip的5倍,而解压的速度相当于gzip的两倍。lzo文件可以根据blockboundaries来进行分块,比如一个1.1G的lzo压缩文件,那么处理第二个 128MBblock的mapper就必须能够确认下一个block的boundary,以便进行解压操作。lzo并没有写什么数据头来做到这一点,而是实现了一个lzoindex文件,将这个文件(foo.lzo.index)写在每个foo.lzo文件中。这个index文件只是简单的包含了每个 block在数据中的offset,这样由于offset已知的缘故,对数据的读写就变得非常的快。通常能达到90-100MB/秒,也就是10-12秒就能读完一个GB的文件。一旦该index文件被创建,任何基于lzo的压缩文件就能通过load该index文件而进行相应的分块,并且一个block 接一个block的被读取。也因此,各个mapper都能够得到正确的block,这就是说,可以只需要进行一个LzopInputStream的封装,就可以在hadoop的mapreduce中并行高效的使用lzo。如果现在有一个job的InputFormat是TextInputFormat,那么就可以用lzop来压缩文件,确保它正确的创建了index,将TextInputFormat换成LzoTextInputFormat,然后job 就能像以前一样正确的运行,并且更加的快。有时候,一个大的文件被lzo压缩过之后,甚至都不用分块就能被单个mapper高效的处理了。

  在hadoop集群中安装lzo

  要在hadoop中搭建lzo使用环境非常简单:

  安装lzop native libraries

  例如:sudo yum install lzop lzo2

  从如下地址下载 hadooplzo支持到源代码:http://github.com/kevinweil/hadoop-lzo

  编译从以上链接checkout下来到代码,通常为:ant compile-native tar

  将编译出来到hadoop-lzo-*.jar部署到hadoop集群到各个slave到某个有效目录下,如$HADOOOP_HOME/lib

  将以上编译所得到hadoop-lzo native libbinary部署到集群到某个有效目录下,如$HADOOP_HOME/lib/native/Linux-amd64-64。

  将如下配置到 core-site.xml 中:

 io.compression.codecsorg.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.BZip2Codec,com.hadoop.compression.lzo.LzoCodec,com.hadoop.compression.lzo.LzopCodec

  io.compression.codec.lzo.class

  com.hadoop.compression.lzo.LzoCodec

  将如下配置到mapred-site.xml中:

  mapred.child.env

  JAVA_LIBRARY_PATH=/path/to/your/native/hadoop-lzo/libs

  如果想要mapreduce再写中间结果时也使用压缩,可以将如下配置也写入到mapred-site.xml中。

  mapred.map.output.compression.codec

  com.hadoop.compression.lzo.LzoCodec

  如果以上所有操作都成功,那么现在就可以尝试使用lzo了。比如打包一个lzo都压缩文件,如lzo_log文件,上传到hdfs中,然后用以下命令进行测试:

  hadoop jar /path/to/hadoop-lzo.jarcom.hadoop.compression.lzo.LzoIndexerhdfs://namenode:9000/lzo_logs

  如果要写一个job来使用lzo,可以找一个job,例如wordcount,将当中到TextInputFormat修改为LzoTextInputForma,其他都不用修改,job就能从hdfs上读入lzo压缩文件,进行分布式都分块并行处理。

  实例:

  1.建立flow表:

  use gj;

  create table top_flow

  (

  src_address string,

  dst_address string,

  src_port int,

  dst_port int,

  trans_protocol int,

  packets bigint,

  bytes bigint,

  flags string,

  start_time timestamp,

  duration double,

  end_time timestamp,

  sensor string

  )

  partitioned by (dd int,hh int,protocol int)

  ROW FORMAT DELIMITED FIELDS TERMINATED BY '|';

  --location 'hdfs://Hadoop:8020/user/hive/warehouse/gj.db/top_flow/'

  生成top_flow的分区建立语句:

  hdfs://Hadoop:8020/user/hive/warehouse/gj.d 777

  hdfs://Hadoop:8020/user/hive/warehouse/gj.db/top_flow/dd=20140707/hh=1/protocol=6/ip=192.168.1.1/

  [Hadoop:21000] > desc formatted top_flow;

  hdfs://Hadoop:8020/user/hive/warehouse/gj.db/sim_event/dd=20140707/hh=1/devtype=TopIdp/ip=192.168.1.1/logtype=IPSAV/

  create table sim_event

  (

  EVENT_ID

  )

  ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

  添分区:

  alter table top_flow add partition(dd=20140707,hh=01,protocol=6);

  hdfs dfs -chmod -R a+w hdfs://hadoop:8020/user/hive/warehouse/gj.db/top_flow/dd=20140707/hh=1/protocol=6

  新建结构一样的表:

  create table top_flow1 like top_flow;

  添加分区:

  alter table top_flow1 add partition(dd=20140707,hh=01,protocol=6);

  alter table top_flow1 partition (dd=20140707,hh=1,protocol=6) set fileformat parquet;

  插入数据

  insert into table top_flow1 partition (dd=20140707,hh=1,protocol=6) select src_address,dst_address,src_port,dst_port,trans_protocol,packets,bytes,flags,start_time,duration,end_time,sensor from top_flow where dd=20140707 and hh=01 and protocol=17;

  删除源表的数据

  alter table top_flow drop partition (dd=20140707,hh=1,protocol=6)

  或者传输文件:

  ----重要内容压缩表:

  1.首先在本地(光盘)安装lzo包,然后安装

  yum install hadoop-lzo-cdh4

  --文件需要拷贝到多个系统的native目录下

  yum install impala-lzo

  2.编辑core-site.xml(注意传输文件到指定目录)

  添加:

  io.compression.codecs

  org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.GzipCodec,

  org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.DeflateCodec,

 

  org.apache.hadoop.io.compress.SnappyCodec,com.hadoop.compression.lzo.LzopCodec

 

  拷贝native文件!!!

  cp /usr/lib/hadoop/lib/native ${Hadop_Home}/lib/

  重启mapreduce和impala服务

  CREATE TABLE top_flow1 (

  src_address string,

  dst_address string,

  src_port int,

  dst_port int,

  trans_protocol int,

  packets bigint,

  bytes bigint,

  flags string,

  start_time timestamp,

  duration double,

  end_time timestamp,

  sensor string)

  partitioned by (dd int,hh int,protocol int)

  STORED AS

  INPUTFORMAT 'com.hadoop.mapred.DeprecatedLzoTextInputFormat'

  OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat';

  insert into table top_flow1 partition (dd=20140707,hh=1,protocol=6) select src_address,dst_address,src_port,dst_port,trans_protocol,packets,bytes,flags,start_time,duration,end_time,sensor from top_flow where dd=20140707 and hh=1 and protocol=6;

  建立索引,文件分块

  hadoop jar /usr/lib/hadoop/lib/hadoop-lzo-cdh4-0.4.15-gplextras.jar com.hadoop.compression.lzo.DistributedLzoIndexer hdfs://hdfs:8020/user/hive/warehouse/gj.db/top_flow1/dd=20140707/hh=1/protocol=6/000000_0.lzo

  [localhost:21000] > invalidate metadata;

  0





收藏 推荐 打印 | 录入:Cstor | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款