数据完整性
检测数据是否损坏的常见措施是：在数据第一次引入系统时，计算校验和（checksum）并在数据通过一个不可靠的通道进行传输时，再次计算校验和。
常见的错误检测码是CRC-32（32位循环冗余校验）
HDFS的数据完整性
datanaode会在一个后台线程中运行一个DataBlockScanner，定期验证存储在这个datanode上的所有数据块。
由于HDFS存储着每个数据块的复本，因此它可以通过数据复本来修复损坏的数据块。
压缩
文件压缩的好处：减少存储文件所需要的磁盘空间，并加速数据在网络和磁盘上的传输。
Codec是压缩-解压算法的一种实现，可以使用CompressionCodec对数据进行压缩和解压缩：
- 对写入输出数据流的数据进行压缩：createOutputStream（OutputStream out），该方法在底层的数据流中对需要以压缩格式写入在此之前尚未压缩的数据新建一个CompressionOutputStream对象。
- 对输入数据流中读取的数据进行解压缩：createInputStream（InputStream in），返回CompressionInputStream对象。
CompressionOutputStream和CompressionInputStream类似于java.util.zip.DeflaterOutputStream和java.util.zip.DeflaterInputStream，只是前者能够重置其底层的压缩或解压缩的方法。
范例：该程序压缩从标准输入读取的数据，然后将其写到标准输出
public class StreamCompressor{
public class void main(String[] args)throws Exception{
String codecClassname = args[0];
Class<?> codecClass = Class.forName(codecClassName);
Configuration conf = new Configuration();
CompressionCodec codec = (CompressionCodec) RefectionUtils.newInstance(condecClass,conf);
ComressionOutputStream out = codec.createOutputStream(System.out);
IOUtils.copyByte(System.in,out,4096,false);
out.finish();
}
}
在MapReduce中使用压缩：
范例：对查找最高气温作业所产生的输出进行压缩：
// cc MaxTemperatureWithCompression Application to run the maximum temperature job producing compressed output
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
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;
//vv MaxTemperatureWithCompression
public class MaxTemperatureWithCompression {
public static void main(String[] args) throws Exception {
if (args.length != 2) {
System.err.println("Usage: MaxTemperatureWithCompression <input path> " +
"<output path>");
System.exit(-1);
}
Job job = new Job();
//驱动程序
job.setJarByClass(MaxTemperature.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);/*]*/
job.setMapperClass(MaxTemperatureMapper.class);
job.setCombinerClass(MaxTemperatureReducer.class);
job.setReducerClass(MaxTemperatureReducer.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
//^^ MaxTemperatureWithCompression
序列化
序列化（serialization）是指将结构化对象转化为字节流以便在网络上传输或写到磁盘进行永久存储的过程。
反序列化（deserialization）是指将字节流转回结构化对象的逆过程。
序列化用于分布式数据处理的两大领域：进程间通信和永久存储。
在Hadoop中，系统中多个节点上进程间通信时通过RPC实现的，RPC协议将消息序列化成二进制流后发送到远程节点，远程节点接着将二进制流反序列化为原始消息。RPC序列化格式如下：
- 紧凑：紧凑格式能充分利用网络带宽
- 快速：进程间通信形成了分布式系统的骨架，所有需要尽量减少序列化和反序列化的性能开销。
- 可扩展：在控制客户端和服务端的过程中，需要直接引进相应的协议。
- 支持互操作：支持以不同语言写的客户端与服务器交互。
Writable接口
Writable接口定义了两个方法：
- 将其状态写入DataOutput二进制流，
- 从DataInput二进制流读取状态
package org.apache.hadoop.io;
import java.io.DataOutput;
import java.io.DataInput;
import java.io.IOException;
public interface Writable{
void write(DataOutput out)throws IOException;
void readFields(DataInput in)throws IOException;
...
}
Wtitable类
Java基本类型的Writable类
基本类型.png
Text类型
ext可以认为是java.lang.String的Writable等价，Text类使用整型来存储字符串编码中所需的字节数，因此最大值为2G，text使用utf-8编码，与String的区别如下：
- 索引：由于Text使用utf-8编码，因此Text类的索引是根据编码后字节序列中的位置来实现的，并非字符串中的Unicode字符，也不是Java char的编码单元。Text的charAt()方法返回的是一个表示Unicode编码位置的int类型值，而String返回一个char类型值，Text的find()方法类似于String的indexOf()方法。
-
可变性：Text是可变的（所有的Hadoop的Writable接口都是可变的，NullWritable除外，它是单例对象），可以通过调用其中的set方法来重用text实例。
类图：
Writable类图.png
参考资料：《Hadoop权威指南》
网友评论