Hadoop大数据处理技术基础与实践（第3版）课件 第6章-Hadoop I-O 操作

第6章HadoopIO操作1了解什么是数据完整性掌握基本的基于文件的数据结构了解常用的压缩算法理解序列化基本原理学习目标28/21/2024参考书38/21/2024

由于每个磁盘或者网络上的I/O操作可能会对正在读写的数据不慎引入错误，如果通过的数据流量非常大，数据发生损坏的几率很高。

检查损坏数据的常用方法是在第一次进入系统时计算数据的校验和，然后只要数据不是在一个可靠的通道上传输，就可能会发生损坏。如果新生成的校验和不完全匹配原始的校验和，那么数据就会被认为是损坏的。

一个常用的错误检测代码是CRC-32(cyclicredundancycheck,循环冗余检查)，计算一个32位的任何大小输入的整数校验和。HDFS数据完整性48/21/2024HDFS的数据完整性58/21/2024HDFS以透明方式校验所有写入它的数据，并在默认设置下，会在读取数据时验证校验和。针对数据的每个io.bytes.per.checksum(默认512字节)字节，都会创建一个单独的校验和。数据节点负责在存储数据及其校验和之前验证它们收到的数据。从客户端和其它数据节点复制过来的数据。客户端写入数据并且将它发送到一个数据节点管线中，在管线的最后一个数据节点验证校验和。HDFS的数据完整性68/21/2024客户端读取数据节点上的数据时，会验证校验和，将其与数据节点上存储的校验和进行对比。每个数据节点维护一个连续的校验和验证日志，因此它知道每个数据块最后验证的时间。每个数据节点还会在后台线程运行一个DataBlockScanner（数据块检测程序），定期验证存储在数据节点上的所有块，为了防止物理存储介质中位衰减锁造成的数据损坏HDFS的数据完整性78/21/2024HDFS的数据完整性88/21/2024HDFS通过复制完整的副本来产生一个新的，无错的副本来“治愈”哪些出错的数据块。工作方式：如果客户端读取数据块时检测到错误，抛出ChecksumException前报告该坏块以及它试图从名称节点中药读取的数据节点。名称节点将这个块标记为损坏的，不会直接复制给客户端或复制该副本到另一个数据节点。它会从其他副本复制一个新的副本。本地文件系统98/21/2024也可能禁用校验和：底层文件系统原生支持校验和。这里通过RawLocalFileSystem来替代LocalFileSystem完成。要在一个应用中全局使用，只需要设置fs.file.impl值为org.apache.hadoop.fs.RawLocalFileSystem来重新map执行文件的URL。或者只想对某些读取禁用校验和校验。

Configurationconf=...

FileSystemfs=newRawLocalFileSystem();

fs.initialize(null,conf);ChecksumFileSystem108/21/2024LocalFileSystem使用ChecksumFileSystem(校验和文件系统)为自己工作，这个类可以很容易添加校验和功能到其他文件系统中。因为ChecksumFileSystem也包含于文件系统中。

FileSystemrawFs=...

FileSystemchecksummedFs=newChecksumFileSystem(rawFs);Hadoop的HDFS和MapReduce子框架主要是针对大数据文件来设计的，在小文件的处理上不但效率低下，而且十分消耗内存资源。解决办法通常是选择一个容器，将这些小文件包装起来，将整个文件作为一条记录，可以获得更高效率的存储和处理，避免了多次打开关闭流耗费计算资源。HDFS提供了两种类型的容器，分别是SequenceFile和MapFile。基于文件的数据结构118/21/2024SequenceFile存储SequenceFile的存储类似于日志文件，所不同的是日志文件的每条记录都是纯文本数据，而SequenceFile的每条记录是可序列化、可持久化的键值数据结构。SequenceFile提供相应的读写器和排序器，写操作根据压缩的类型分为3种。（1）Writer：无压缩写数据。（2）RecordCompressWriter：记录级压缩文件，只压缩值。（3）BlockCompressWrite：块级压缩文件，键值采用独立压缩方式。读取操作实际上可以读取上述3种类型。128/21/2024SequenceFile存储在存储结构上，SequenceFile主要由一个Header后跟多条Record组成，如图所示。138/21/2024SequenceFile存储当保存的记录有很多的时候，可以把一连串的记录组织到一起，统一压缩成一个块。148/21/2024用命令行接口显示序列文件

使用-text选项显示文本格式的序列文件。

%hadoopfs-textnumber.seq158/21/2024MapFile是排序后的SequenceFile，并且它会额外生成一个索引文件提供按键的查找。读写MapFile与读写SequenceFile非常类似，只需要换成MapFie.Reader和MapFile.Writer就可以了。在命令行显示MapFile的文件内容同样要用-text。MapFile168/21/2024MapFile是排序后的SequenceFile，并且它会额外生成一个索引文件提供按键的查找。读写MapFile与读写SequenceFile非常类似，只需要换成MapFie.Reader和MapFile.Writer就可以了。在命令行显示MapFile的文件内容同样要用-text。压缩178/21/2024文件压缩188/21/2024文件压缩两大好处：减少存储文件所需要的空间且加快了数据在网络上或从磁盘上或到磁盘上的传输速度。各种压缩算法的压缩比：

编码和解码198/21/2024Codec是coder与decoder的缩略词，实现了一种压缩-解压算法。Hadoop中的压缩与解压的类实现CompressionCodec接口

编码和解码208/21/2024CompressionCodec有两个方法轻松地压缩和解压数据。使用usethe

createOutputStream(OutputStreamout)创建一个CompressionOutputStream，将其以压缩格式写入底层的流。使用createInputStream(InputStreamin)获取一个CompressionInputStream，从底层的流读取未压缩的数据。

编码和解码218/21/202401packagecom.laos.hadoop;0203importorg.apache.hadoop.conf.Configuration;04importorg.apache.hadoop.io.IOUtils;05importpress.CompressionCodec;06importpress.CompressionOutputStream;07importorg.apache.hadoop.util.ReflectionUtils;0809publicclassStreamCompressor{10

publicstaticvoidmain(String[]args)throwsException{11

StringcodecClassname="press.GzipCodec";12

Class<?>codecClass=Class.forName(codecClassname);13

Configurationconf=newConfiguration();14

CompressionCodeccodec=(CompressionCodec)ReflectionUtils15

.newInstance(codecClass,conf);16

//将读入数据压缩至System.out17

CompressionOutputStreamout=codec.createOutputStream(System.out);18

IOUtils.copyBytes(System.in,out,4096,false);19

out.finish();20

}2122}

在unix窗口输入命令：$echo"Test"|hadoopjarhadoop-itest.jarcom.laos.hadoop.StreamCompressor|gunzip编码和解码228/21/2024使用CompressionCodecFactory方法来推断CompressionCodec

在阅读一个压缩文件时，我们可以从扩展名来推断出它的编码/解码器。以.gz结尾的文件可以用GzipCodec来阅读。CompressionCodecFactory提供了getCodec()方法，从而将文件扩展名映射到相应的CompressionCodec。编码和解码238/21/2024

01packagecom.laos.hadoop;0203importjava.io.InputStream;04importjava.io.OutputStream;05import.URI;0607importorg.apache.hadoop.conf.Configuration;08importorg.apache.hadoop.fs.FileSystem;09importorg.apache.hadoop.fs.Path;10importorg.apache.hadoop.io.IOUtils;11importpress.CompressionCodec;12importpress.CompressionCodecFactory;1314publicclassFileDecompressor{15

publicstaticvoidmain(String[]args)throwsException{16

Stringuri=args[0];17

Configurationconf=newConfiguration();18

FileSystemfs=FileSystem.get(URI.create(uri),conf);1920

PathinputPath=newPath(uri);21

CompressionCodecFactoryfactory=newCompressionCodecFactory(conf);22

CompressionCodeccodec=factory.getCodec(inputPath);23

if(codec==null){24

System.err.println("Nocodecfoundfor"+uri);25

System.exit(1);26

}27

StringoutputUri=CompressionCodecFactory.removeSuffix(uri,codec28

.getDefaultExtension());29

InputStreamin=null;30

OutputStreamout=null;31

try{32

in=codec.createInputStream(fs.open(inputPath));33

out=fs.create(newPath(outputUri));34

IOUtils.copyBytes(in,out,conf);35

}finally{36

IOUtils.closeStream(in);37

IOUtils.closeStream(out);38

}39

}40}压缩和输入分隔248/21/2024考虑如何压缩哪些将由MapReduce处理的数据时，考虑压缩格式是否支持分隔很重要。

例如，gzip格式使用default来存储压缩过的数据，default将数据作为一系列压缩过的块存储，但是每块的开始没有指定用户在数据流中的任意点定位到下一个块的起始位置，而是自身与数据同步，所以gzip不支持分隔机制。在MapReduce中使用压缩258/21/2024如果要压缩MapReduce作业的输出，设置press为true，pression.codec属性指定编码解码器。

如果输入的文件时压缩过的，MapReduce读取时，它们会自动解压，根据文件扩展名来决定使用那一个压缩解码器。publicclassMaxTemperatureWithCompression{17

publicstaticvoidmain(String[]args)throwsIOException{18

if(args.length!=2){19

System.err.println("Usage:MaxTemperatureWithCompression<inputpath>"+20

"<outputpath>");21

System.exit(-1);22

}23

24

JobConfconf=newJobConf(MaxTemperatureWithCompression.class);conf.setJobName("Maxtemperaturewithoutputcompression");25

FileInputFormat.addInputPath(conf,newPath(args[0]));26

FileOutputFormat.setOutputPath(conf,newPath(args[1]));27

28

conf.setOutputKeyClass(Text.class);29

conf.setOutputValueClass(IntWritable.class);30

31

conf.setBoolean("press",true);32

conf.setClass("pression.codec",GzipCodec.class,33

CompressionCodec.class);

34

JobClient.runJob(conf);35

}36

}序列化：将结构化对象转换为字节流以便于通过网络进行传输或写入存储的过程。反序列化：将字节流转为一系列结构化对象的过程。

序列化用在两个地方：进程间通信和持久存储。在Hadoop中，节点之间的进程间通信是用远程过程调用(RPC)。RPC协议将使用序列化将消息编码为二进制流(发送到远程节点)，此后在接收端二进制流被反序列化为消息。Hadoop使用自己的序列化格式Writables。序列化268/21/2024Writable接口packageorg.apache.hadoop.io;importjava.io.DataOutput;importjava.io.DataInput;importjava.io.IOException;

publicinterfaceWritable{

voidwrite(DataOutputout)throwsIOException;

//将状态写入二进制格式的流

voidreadFields(DataInputin)throwsIOException;//从二进制格式的流读出其状态}WritableComparable和Comparator

IntWritable实现了WritableComparable接口。而WritableComparable继承了Writable和Comparable。278/21/2024Writabl