在hadoop中使用lzo的压缩算法可以减小数据 的大小和数据的磁盘读写时间，不仅如此，lzo是基于block分块的，这样他就允许数据被分解成chunk，并行的被hadoop处理。这样的特点，就可以让lzo在hadoop上成为一种非常好用的压缩格式。
lzo本身不是splitable的，所以当数据为text格式时，用lzo压缩出来的数据当做job的输入是一个文件作为一个map。但是sequence file本身是分块的，所以sequence file格式的文件，再配上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的划分来split job。不仅如此，lzo在设计时就考虑到了效率问题，它的解压速度是gzip的两倍，这就让它能够节省很多的磁盘读写，它的压缩比的不如gzip，大约 压缩出来的文件比gzip压缩的大一半，但是这样仍然比没有经过压缩的文件要节省20%-50%的存储空间 ，这样就可以在效率上大大的提高job执行的速度。以下是一组压缩对比数据，使用一个8.0GB的未经过压缩的数据来进行对比：
压缩格式 文件 大小(GB) 压缩时间 解压时间
None some_logs 8.0 - -
Gzip some_logs.gz 1.3 241 72
LZO some_logs.lzo 2.0 55 35
可以看出，lzo压缩文件会比gzip压缩文件稍微大一些，但是仍然比原始文件要小很多倍，并且lzo文件压缩的速度几乎相当于gzip的5倍，而解压的 速度相当于gzip的两倍。lzo文件可以根据block boundaries来进行分块，比如一个1.1G的lzo压缩文件，那么处理第二个128MB block的mapper就必须能够确认下一个block的boundary，以便进行解压操作。lzo并没有写什么数据头来做到这一点，而是实现了一个 lzo index文件，将这个文件（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高效的处理了。
【同转LZO算法】
// 本程序采用 Lempel-Ziv 压缩算法, 云风没有对这个算法做深入研究
// 代码是根据 Markus Franz Xaver Johannes Oberhumer 的 LZO 改写
// 而成, 所以算法上的问题请不要问我, 如果你对这个压缩算法有兴趣,
// 请拜访 LZO 的主页
// http://wildsau.idv.uni-linz.ac.at/mfx/lzo.html
// LZO 属于 GNU 软件, 在此声明, 本文件中的代码使用权利不同于风魂游
// 戏程序库中的其他部分, 请遵守 GNU 协议使用.
//#include <stdio.h>
//#include <string.h>
#include <stdlib.h>
#define byte unsigned char
int _stdcall compress(void *src, unsigned src_len, void *dst);
int _stdcall decompress(void *src, unsigned src_len, void *dst);
static unsigned _do_compress (byte *in, unsigned in_len, byte *out, unsigned *out_len)
{
static long wrkmem [16384L];
register byte *ip;
byte *op;
byte *in_end = in + in_len;
byte *ip_end = in + in_len - 13;
byte *ii;
byte **dict = (byte **)wrkmem;
op = out;
ip = in;
ii = ip;
ip += 4;
for(;;)
{
register byte *m_pos;
unsigned m_off;
unsigned m_len;
unsigned dindex;
dindex = ((0x21*(((((((unsigned)(ip[3])<<6)^ip[2])<<5)^ip[1])<<5)^ip[0]))>>5) & 0x3fff;
m_pos = dict [dindex];
if(((unsigned)m_pos < (unsigned)in) ||
(m_off = (unsigned)((unsigned)ip-(unsigned)m_pos) ) <= 0 ||
m_off > 0xbfff)
goto literal;
if(m_off <= 0x0800 || m_pos[3] == ip[3])
goto try_match;
dindex = (dindex & 0x7ff ) ^ 0x201f;
m_pos = dict[dindex];
if((unsigned)(m_pos) < (unsigned)(in) ||
(m_off = (unsigned)( (int)((unsigned)ip-(unsigned)m_pos))) <= 0 ||
m_off > 0xbfff)
goto literal;
if (m_off <= 0x0800 || m_pos[3] == ip[3])
goto try_match;
goto literal;
try_match:
if(*(unsigned short*)m_pos == *(unsigned short*)ip && m_pos[2]==ip[2])
goto match;
literal:
dict[dindex] = ip;
++ip;
if (ip >= ip_end)
break;
continue;
match:
dict[dindex] = ip;
if(ip - ii > 0)
{
register unsigned t = ip - ii;
if (t <= 3)
op[-2] |= (byte)t;
else if(t <= 18)
*op++ = (byte)(t - 3);
else
{
register unsigned tt = t - 18;
*op++ = 0;
while(tt > 255)
{
tt -= 255;
*op++ = 0;
}
*op++ = (byte)tt;
}
do *op++ = *ii++; while (--t > 0);
}
ip += 3;
if(m_pos[3] != *ip++ || m_pos[4] != *ip++ || m_pos[5] != *ip++ ||
m_pos[6] != *ip++ || m_pos[7] != *ip++ || m_pos[8] != *ip++ )
{
--ip;
m_len = ip - ii;
if(m_off <= 0x0800 )
{
--m_off;
*op++ = (byte)(((m_len - 1) << 5) | ((m_off & 7) << 2));
*op++ = (byte)(m_off >> 3);
}
else
if (m_off <= 0x4000 )
{
-- m_off;
*op++ = (byte)(32 | (m_len - 2));
goto m3_m4_offset;
}
else
{
m_off -= 0x4000;
*op++ = (byte)(16 | ((m_off & 0x4000) >> 11) | (m_len - 2));
goto m3_m4_offset;
}
}
else
{
{
byte *end = in_end;
byte *m = m_pos + 9;
while (ip < end && *m == *ip)
m++, ip++;
m_len = (ip - ii);
}
if(m_off <= 0x4000)
{
--m_off;
if (m_len <= 33)
*op++ = (byte)(32 | (m_len - 2));
else
{
m_len -= 33;
*op++=32;
goto m3_m4_len;
}
}
else
{
m_off -= 0x4000;
if(m_len <= 9)
*op++ = (byte)(16|((m_off & 0x4000) >> 11) | (m_len - 2));
else
{
m_len -= 9;
*op++ = (byte)(16 | ((m_off & 0x4000) >> 11));
m3_m4_len:
while (m_len > 255)
{
m_len -= 255;
*op++ = 0;
}
*op++ = (byte)m_len;
}
}
m3_m4_offset:
*op++ = (byte)((m_off & 63) << 2);
*op++ = (byte)(m_off >> 6);
}
ii = ip;
if (ip >= ip_end)
break;
}
*out_len = op - out;
return (unsigned) (in_end - ii);
}
int _stdcall compress(void *in, unsigned in_len,
void *out)
{
byte *op = out;
unsigned t,out_len;
if (in_len <= 13)
t = in_len;
else
{
t = _do_compress (in,in_len,op,&out_len);
op += out_len;
}
if (t > 0)
{
byte *ii = (byte*)in + in_len - t;
if (op == (byte*)out && t <= 238)
*op++ = (byte) ( 17 + t );
else
if (t <= 3)
op[-2] |= (byte)t ;
else
if (t <= 18)
*op++ = (byte)(t-3);
else
{
unsigned tt = t - 18;
*op++ = 0;
while (tt > 255)
{
tt -= 255;
*op++ = 0;
}
*op++ = (byte)tt;
}
do *op++ = *ii++; while (--t > 0);
}
*op++ = 17;
*op++ = 0;
*op++ = 0;
return (op - (byte*)out);
}
int _stdcall decompress (void *in, unsigned in_len,
void *out)
{
register byte *op;
register byte *ip;
register unsigned t;
register byte *m_pos;
byte *ip_end = (byte*)in + in_len;
op = out;
ip = in;
if(*ip > 17)
{
t = *ip++ - 17;
if (t < 4)
goto match_next;
do *op++ = *ip++; while (--t > 0);
goto first_literal_run;
}
for(;;)
{
t = *ip++;
if (t >= 16) goto match;
if (t == 0)
{
while (*ip == 0)
{
t += 255;
ip++;
}
t += 15 + *ip++;
}
* (unsigned *) op = * ( unsigned *) ip;
op += 4; ip += 4;
if (--t > 0)
{
if (t >= 4)
{
do
{
* (unsigned * ) op = * ( unsigned * ) ip;
op += 4; ip += 4; t -= 4;
} while (t >= 4);
if (t > 0) do *op++ = *ip++; while (--t > 0);
}
else
do *op++ = *ip++; while (--t > 0);
}
first_literal_run:
t = *ip++;
if (t >= 16)
goto match;
m_pos = op - 0x0801;
m_pos -= t >> 2;
m_pos -= *ip++ << 2;
*op++ = *m_pos++; *op++ = *m_pos++; *op++ = *m_pos;
goto match_done;
for(;;)
{
match:
if (t >= 64)
{
m_pos = op - 1;
m_pos -= (t >> 2) & 7;
m_pos -= *ip++ << 3;
t = (t >> 5) - 1;
goto copy_match;
}
else
if (t >= 32)
{
t &= 31;
if (t == 0)
{
while (*ip == 0)
{
t += 255;
ip++;
}
t += 31 + *ip++;
}
m_pos = op - 1;
m_pos -= (* ( unsigned short * ) ip) >> 2;
ip += 2;
}
else
if (t >= 16)
{
m_pos = op;
m_pos -= (t & 8) << 11;
t &= 7;
if (t == 0)
{
while (*ip == 0)
{
t += 255;
ip++;
}
t += 7 + *ip++;
}
m_pos -= (* ( unsigned short *) ip) >> 2;
ip += 2;
if (m_pos == op)
goto eof_found;
m_pos -= 0x4000;
}
else
{
m_pos = op - 1;
m_pos -= t >> 2;
m_pos -= *ip++ << 2;
*op++ = *m_pos++; *op++ = *m_pos;
goto match_done;
}
if (t >= 6 && (op - m_pos) >= 4)
{
* (unsigned *) op = * ( unsigned *) m_pos;
op += 4; m_pos += 4; t -= 2;
do
{
* (unsigned *) op = * ( unsigned *) m_pos;
op += 4; m_pos += 4; t -= 4;
}while (t >= 4);
if (t > 0)
do *op++ = *m_pos++; while (--t > 0);
}
else
{
copy_match:
*op++ = *m_pos++; *op++ = *m_pos++;
do *op++ = *m_pos++; while (--t > 0);
}
match_done:
t = ip[-2] & 3;
if (t == 0) break;
match_next:
do *op++ = *ip++; while (--t > 0);
t = *ip++;
}
}
eof_found:
if (ip != ip_end) return -1;
return (op - (byte*)out);
}
分享到：
相关推荐
《Hadoop LZO压缩工具详解》 在大数据处理领域，Hadoop是一个不可或缺的开源框架，它为海量数据的存储和处理提供了强大支持。而在数据压缩方面，Hadoop LZO是其中一种常用的数据压缩工具，本文将详细介绍这个工具...
《Hadoop与LZO压缩：深入理解hadoop-lzo-0.4.15.tar.gz》 在大数据处理领域，Hadoop是不可或缺的核心组件，它为海量数据的存储和计算提供了分布式解决方案。而LZO（Lempel-Ziv-Oberhumer）是一种高效的无损数据压缩...
总的来说，Hadoop-LZO是一个针对Hadoop的LZO压缩解决方案，它提供了一种高效的数据压缩方式，尤其适合大数据处理场景，通过这个特定版本的库，开发者可以在Mac环境下为Hadoop 2.8.0实现LZO压缩功能，提升数据处理的...
hadoop支持LZO压缩配置 将编译好后的hadoop-lzo-0.4.20.jar 放入hadoop-2.7.2/share/hadoop/common/ core-site.xml增加配置支持LZO压缩 <name>io.compression.codecs org.apache.hadoop.io....
Hadoop-LZO是一款专门为Apache Hadoop设计的高效数据压缩库，它基于开源的LZO压缩算法，旨在提高Hadoop生态系统中的数据存储和处理效率。在大数据领域，压缩技术是至关重要的，因为它可以显著减少存储需求，加快数据...
- `hadoop-lzo-0.4.20.jar`：这是一个已经编译好的Hadoop-LZO库，可以直接添加到Hadoop的类路径中，以便在不修改Hadoop源码的情况下使用LZO压缩。 - `hadoop-lzo-master.zip`：这是Hadoop-LZO项目的源代码，用户可以...
【标题】"lzo 2.0.6、hadoop-lzo-master、apache-maven" 涉及的主要是三个关键元素：LZO压缩库、Hadoop-LZO项目以及Apache Maven，这些都是在大数据处理和软件构建领域的重要工具。 【LZO 2.0.6】： LZO（Lempel-...
在Hadoop中，LZO压缩可以提高数据存储效率，降低存储成本，同时在MapReduce作业中，通过预压缩数据，可以减少网络传输的负担，提高集群性能。 标题提到的“hadoop-lzo所需包”指的是为了在Hadoop环境中使用LZO压缩...
hadoop配置支持LZO压缩必备，版本号hadoop-lzo-0.4.20-SNAPSHOT.jar，
而“hadoop-lzo-master.zip”则是Hadoop-LZO项目的源代码，通常包含了项目的所有文件，包括Java源代码、构建脚本和文档，用户可以下载并根据指导进行编译和部署，以在Hadoop集群上启用LZO压缩功能。 在大数据处理中...
hadoop lzo 压缩jar包，本人已经编译好，提供大家下载。
Hadoop与LZO压缩 Hadoop是一个开源框架，主要用于处理和存储大规模数据，它由Apache软件基金会开发。在大数据处理领域，Hadoop以其分布式计算模型（MapReduce）和可扩展性而闻名。为了提高数据存储和传输效率，...
Hadoop-lzo是针对Hadoop生态系统优化的Lzo实现，它允许Hadoop集群在MapReduce作业中直接处理Lzo压缩的文件。Hadoop-lzo不仅包含了对Lzo库的集成，还提供了一个名为`GzipIndexer`的工具，用于创建Lzo文件的索引，使得...
将生成的 build/hadoop-lzo-0.4.15.jar cp 到 /usr/local/hadoop-1.0.2/lib 测试解压程序 bin/hadoop jar /usr/local/hadoop-1.0.2/lib/hadoop-lzo-0.4.15.jar ...