|
1.数据的完整性 1).HDFS的数据完整性
2).LocalFileSystem Hadoop的LocalFileSystem执行客户端的校验和验证(原理见权威指南)，当检测到错误，LocalFileSystem将抛出一个ChecksumException 如果想针对一些操作禁用校验和 下面给出示例： Configuration conf = ... FileSystem fs = new RawLocalFileSystem(); fs.initialize(null, conf);
3).ChecksumFileSystem LocalFileSystem通过来完成任务，有了ChecksumFileSystem，向其他文件(无校验和系统)加入校验和就非常简单，因为ChecksumFileSystem类继承自FileSystem，一般用法： FileSystem rawFs = ... FileSystem checksummedFs = new ChecksumFileSystem(rawFs);
2.压缩 文件压缩有两大好处：可以减少存储文件所需要的磁盘空间；可以加速数据在网络和磁盘的传输。有很多不同的压缩格式、工具和算法，如下表压缩格式总结：
1)codec压缩 Hadoop的压缩codec：
通过CompressionCodec对数据进行压缩和解压缩 对于输出流的数据压缩，可用createOutputStream(OutputStream out)对尚未压缩对数据创建一个CompressionOutputStream对象，相反，对于输出流的解压缩，则调用createInputStream(InputStream in)获取CompressionInputStream。 下面程序显示了程序压缩从标准输入读取的数据，然后将其写到标准输出：
public class StreamCompressor { public static void main(String[] args) throws Exception {
String codecClassname = args[0];
Class<?> codecClass = Class.forName(codecClassname);
Configuration conf = new Configuration();
CompressionCodec codec = (CompressionCodec) ReflectionUtils
.newInstance(codecClass, conf);
CompressionOutputStream out = codec.createOutputStream(System.out);
IOUtils.copyBytes(System.in, out, 4096, false);
out.finish();
}
}
通过CompressionCodecFactory判断CompressionCodec 通过使用其getCodec()方法，CompressionCodecFactory提供了一种方法可以将文件扩展名映射名映射一个CompressionCodec 下面程序使用这个特性来对文件进行解压缩：
public class FileDecompressor { public static void main(String[] args) throws Exception {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path inputPath = new Path(uri);
CompressionCodecFactory factory = new CompressionCodecFactory(conf);
CompressionCodec codec = factory.getCodec(inputPath); if (codec == null) {
System.err.println("No codec found for " + uri);
System.exit(1);
}
String outputUri = CompressionCodecFactory.removeSuffix(uri,
codec.getDefaultExtension());
InputStream in = null;
OutputStream out = null; try {
in = codec.createInputStream(fs.open(inputPath));
out = fs.create(new Path(outputUri));
IOUtils.copyBytes(in, out, conf);
} finally {
IOUtils.closeStream(in);
IOUtils.closeStream(out);
}
}
}
原生类库 使用原生类库可以提高性能，请注意并不是所有的实现都提供了本地库，但是有的实现却只提供了原生类库（native实现,native implementation）。下面是几种实现以及原生类库情况：
如果使用原生类库并且在应用中执行大量的压缩和解压缩操作，可以考虑使用CodecPool 下面程序使用压缩池对读取字标准输入的数据进行压缩，然后将其写到标准输出
public class PooledStreamCompressor { public static void main(String[] args) throws Exception {
String codecClassname = args[0];
Class<?> codecClass = Class.forName(codecClassname);
Configuration conf = new Configuration();
CompressionCodec codec = (CompressionCodec);
ReflectionUtils.newInstance(codecClass, conf);
Compressor compressor = null; try {
compressor = CodecPool.getCompressor(codec);
CompressionOutputStream out = codec.createOutputStream(System.out,
compressor);
IOUtils.copyBytes(System.in, out, 4096, false);
out.finish();
} finally {
CodecPool.returnCompressor(compressor);
}
}
}
3.压缩和输入分片 当考虑将怎样压缩将被MapReduce处理的数据的时候，考虑一下压缩格式是否支持split非常重要。
怎么挑选压缩格式，对于巨大的、没有存储边界的文件、如日志文件，可以考虑如下选项：
4.在MapReduce中使用压缩 如果在MapReduce输出中使用压缩操作，应在作业配置过程中将mapred.ouput.compress属性设为true和mapred.ouput.compression.codec属性设置为打算使用的压缩codec的类名(如上面的图4-4) 下面的程序对查找最高气温作业所产生输出进行压缩
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);
}
}
实际上也可以对map阶段的中间输入进行压缩，也可以提供网络中间传输性能，启用map任务输出压缩和设置压缩格式的配置属性如下图
下面是用新的API进行设置启动gzip压缩map输出
Configuration conf = new Configuration();
conf.setBoolean("mapred.compress.map.output", true);
conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class);
Job job = new Job(conf);
5.序列化 序列化是指将结构化的对象转为字节流以，便于通过网络进行传输或者写入持久存储的过程。反序列化则是将字节流转为一系列结构化对象的过程。 序列化在分布式数据处理的两大领域经常出现：进程间通信和永久存储。 在Hadoop中，提供了Hadoop RPC + 序列化的通信方式，通常RPC序列化格式特点有：紧凑、快速、可扩展、互操作
1).Writable接口 Hadoop使用自己的序列化格式Writable，它的格式紧凑、速度快，是Hadoop的核心，Writable接口定义了两个方法：一个将其状态写到DataOutput二进制流，另一个从DataInput二进制流读取其状态： 下面程序实例Writable类的具体用途 首先序列化：
IntWritable writable = new IntWritable();
writable.set(163); //或者IntWritable writable = new IntWritable(163); public static byte[] serialize(Writable writable) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputStream dataOut = new DataOutputStream(out);
writable.write(dataOut);
dataOut.close(); return out.toByteArray();
} byte[] bytes = serialize(writable); //调用 assertThat(bytes.length, is(4));
然后反序列化：
public static byte[] deserialize(Writable writable, byte[] bytes) throws IOException {
ByteArrayInputStream in = new ByteArrayInputStream(bytes);
DataInputStream dataIn = new DataInputStream(in);
writable.readFields(dataIn);
dataIn.close(); return bytes;
}
IntWritable newWritable = new IntWritable();
deserialize(newWritable, bytes);
assertThat(newWritable.get(), is(163));
2).Writable类 A.Java基本类型的Writable封装器 下图中，Writable类对Java所有封装类都包含有get和set方法：
Java基本类型Writable类,如下图
B.Text类型 索引：对Text类的索引是根据编码后字节序列中的位置实现 ，并非字符串中的Unicode字符，也不是Java Char的编码单元 Text的CharAt方法用法如下所示：
Text t = new Text("hadoop");
assertThat(t.getLength(), is(6));
assertThat(t.getBytes().length, is(6));
assertThat(t.charAt(2), is((int) 'd'));
assertThat("Out of bounds", t.charAt(100), is(-1));
Unicode：一旦开始使用需要多个字节编码的字符的时候，Text和String之间的区别就很明显的了，下面程序测试Text和String的不同：
public class StringTextComparisonTest {
@Test public void string() throws UnsupportedEncodingException {
String s = "\u0041\u00DF\u6771\uD801\uDC00";
assertThat(s.length(), is(5));
assertThat(s.getBytes("UTF-8").length, is(10));
assertThat(s.indexOf("\u0041"), is(0));
assertThat(s.indexOf("\u00DF"), is(1));
assertThat(s.indexOf("\u6771"), is(2));
assertThat(s.indexOf("\uD801\uDC00"), is(3));
assertThat(s.charAt(0), is('\u0041'));
assertThat(s.charAt(1), is('\u00DF'));
assertThat(s.charAt(2), is('\u6771'));
assertThat(s.charAt(3), is('\uD801'));
assertThat(s.charAt(4), is('\uDC00'));
assertThat(s.codePointAt(0), is(0x0041));
assertThat(s.codePointAt(1), is(0x00DF));
assertThat(s.codePointAt(2), is(0x6771));
assertThat(s.codePointAt(3), is(0x10400));
}
@Test public void text() {
Text t = new Text("\u0041\u00DF\u6771\uD801\uDC00");
assertThat(t.getLength(), is(10));
assertThat(t.find("\u0041"), is(0));
assertThat(t.find("\u00DF"), is(1));
assertThat(t.find("\u6771"), is(3));
assertThat(t.find("\uD801\uDC00"), is(6));
assertThat(t.charAt(0), is(0x0041));
assertThat(t.charAt(1), is(0x00DF));
assertThat(t.charAt(3), is(0x6771));
assertThat(t.charAt(6), is(0x10400));
}
}
上面程序中String的长度是其所含char编码的单元个数(5，前三个字符和最后一个代理对)，但是Text对象的长度确是其UTF-8编码的字节数(10=1+2+3+4)
迭代：将Text对象变成java.io.ByteBuffer，然后对缓冲的Text反复调用bytesToCodePoint（）静态方法。这个方法提取下一个代码点作为int然后更新缓冲中的位置。当bytesToCodePoint（）返回-1时，检测到字符串结束。下面程序遍历Text对象中的字符：
public class TextIterator { public static void main(String[] args) {
Text t = new Text("\u0041\u00DF\u6771\uD801\uDC00");
ByteBuffer buf = ByteBuffer.wrap(t.getBytes(), 0, t.getLength()); int cp; while (buf.hasRemaining() && (cp = Text.bytesToCodePoint(buf)) != -1) {
System.out.println(Integer.toHexString(cp));
}
}
}
易变性：Text相比String是可变的，可以通过调用其中一个set()方法来重用Text实例：
C.BytesWritable BytesWritable是对二进制数据数组的封装。它的序列化格式为一个用于指定后面数据字节数的整数域（4字节），后跟字节本身。例如，长度为2的字节数组包含数值3和5，序列化形式为一个4字节整数（00000002）和该数组中的两个字节（03）和（05）
D.NullWritable NullWritable是Writable的一个特殊类型。它的序列化长度为0，它并不从数据流中读取数据，也不写入数据。它充当占位符；例如，在MapReduce中，如果不需要使用键或值，就可以将键或值声明为NullWritable——结果是存储常量控制。它是一个可变的单实例类型：通过调用NullWritable.get()方法可以获取这个实例。
E.ObjectWritable和GenericWritable
F.Writable集合类 org.apache.hadoop.io包中有四种Writable集合类型，分别是ArrayWritable，TwoDArrayWritable，MapperWritable和SortedMapWritable：
6.定制Writable类型程序 1).存储一对Text对象的Writable类型
import org.apache.hadoop.io.*; public class TextPair implements WritableComparable { private Text first; private Text second; public TextPair() {
set(new Text(), new Text());
} public TextPair(String first, String second) {
set(new Text(first), new Text(second));
} public TextPair(Text first, Text second) {
set(first, second);
} public void set(Text first, Text second) { this.first = first; this.second = second;
} public Text getFirst() { return first;
} public Text getSecond() { return second;
}
@Override public void write(DataOutput out) throws IOException {
first.write(out);
second.write(out);
}
@Override public void readFields(DataInput in) throws IOException {
first.readFields(in);
second.readFields(in);
}
@Override public int hashCode() { return first.hashCode() * 163 + second.hashCode();
}
@Override public boolean equals(Object o) { if (o instanceof TextPair) {
TextPair tp = (TextPair) o; return first.equals(tp.first) && second.equals(tp.second);
} return false;
}
@Override public String toString() { return first + "\t" + second;
}
@Override public int compareTo(TextPair tp) { int cmp = first.compareTo(tp.first); if (cmp != 0) { return cmp;
} return second.compareTo(tp.second);
}
}
2).比较TextPair字节表示RawComparator
public static class Comparator extends WritableComparator { private static final Text.Comparator TEXT_COMPARATOR = new Text.Comparator(); public Comparator() { super(TextPair.class);
}
@Override public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { try { int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1); int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2); int cmp = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2); if (cmp != 0) { return cmp;
} return TEXT_COMPARATOR.compare(b1, s1 + firstL1, l1 - firstL1,
b2, s2 + firstL2, l2 - firstL2);
} catch (IOException e) { throw new IllegalArgumentException(e);
}
}
} static {
WritableComparator.define(TextPair.class, new Comparator());
}
3).定制的RawComparator用于比较TextPair对象字节表示的第一个字段
public static class FirstComparator extends WritableComparator { private static final Text.Comparator TEXT_COMPARATOR = new Text.Comparator(); public FirstComparator() { super(TextPair.class);
}
@Override public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { try { int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1); int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2); return TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
} catch (IOException e) { throw new IllegalArgumentException(e);
}
}
@Override public int compare(WritableComparable a, WritableComparable b) { if (a instanceof TextPair && b instanceof TextPair) { return ((TextPair) a).first.compareTo(((TextPair) b).first);
} return super.compare(a, b);
}
}
序列化框架和Avro详见权威指南
7.基于文件的数据结构 1).SequenceFile 如果想记录二进制类型，纯文本是不合适的，这种情况下Hadoop的SequenceFile类非常合适，因为它提供了二进制键/值对的永久存储的数据结构。SequenceFiles同样也可以作为小文件的容器。而HDFS和MapReduce是针对大文件进行优化的，所以通过SequenceFile类型将小文件包装起来，可以获得更高效的存储和处理
SequenceFile的写操作 通过creatWriter()静态方法可以创建SequenceFile对象，并返回SequenceFile.Writer实例。该静态方法有多个重载版本，但都需要指定待写入的数据流（FSDataOutputStream或FileSystem或Path），Configuration对象，以及键和值的类型。另外可选参数包括压缩类型以及相应的codec，Progressable回调函数用于通知写入的进度，以及在SequenceFile头文件中存储的Metadata实例。存储在SequenceFile中的键和值并不一定需要Writable类型，任何可以通过Serialization类实现序列化和反序列化的类型均可以。一旦拥有SequenceFile.Writer实例，就可以通过append()方法在文件末尾附加键/值。写完后调用close()： 下面的程序写入SequenceFile对象：
public class SequenceFileWriteDemo { private static final String[] DATA = { "One, two, buckle my shoe", "Three, four, shut the door", "Five, six, pick up sticks", "Seven, eight, lay them straight", "Nine, ten, a big fat hen" }; public static void main(String[] args) throws IOException {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path(uri);
IntWritable key = new IntWritable();
Text value = new Text();
SequenceFile.Writer writer = null; try {
writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),
value.getClass()); for (int i = 0; i < 100; i++) {
key.set(100 - i);
value.set(DATA[i % DATA.length]);
System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key,
value);
writer.append(key, value);
}
} finally {
IOUtils.closeStream(writer);
}
}
}
读取SequenceFile 从头到尾读取顺序文件的过程是创建SequenceFile.Reader实例后反复调用next()方法迭代读取记录的过程。读取的是哪条记录与你使用的序列化框架相关。如果你使用的是Writable类型，那么通过键和值作为参数的next()方法可以将数据流中的下一条键值对读入变量中： public boolean next(Writable key, Writable val); 如果读取成功则返回true，如果以读到文件尾则返回false。如果读取非Writable类型的序列化框架，则需要使用 public Object next( Object key ) throws IOException; public Object getCurrentValue(Object val) throws IOException; 这种情况下请确保在io.serializations属性已经设置了你想使用的序列化框架。如果next()方法返回非空对象，则可以从数据流中读取键值对，并且可以通过getCurrentValue()方法读取该值。否则返回null表示到文件尾： 读取SequenceFile示例：
public class SequenceFileReadDemo { public static void main(String[] args) throws IOException {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path path = new Path(uri);
SequenceFile.Reader reader = null; try {
reader = new SequenceFile.Reader(fs, path, conf);
Writable key = (Writable)
ReflectionUtils.newInstance(reader.getKeyClass(), conf);
Writable value = (Writable)
ReflectionUtils.newInstance(reader.getValueClass(), conf); long position = reader.getPosition(); while (reader.next(key, value)) {
String syncSeen = reader.syncSeen() ? "*" : "";
System.out.printf("[%s%s]\t%s\t%s\n", position, syncSeen, key, value);
position = reader.getPosition(); // beginning of next record }
} finally {
IOUtils.closeStream(reader);
}
}
}
在顺序文件中搜索给定位置有两种方法：seek()和同步点找到记录便捷 通过命令行接口显示SequenceFile对象：hadoop fs -text ，可以以文本形式显示顺序文件的内容。该选项可以识别gzip压缩的文件和顺序文件，否则假设输入为纯文本文件。
MapReduce是对多个顺序文件进行排序（合并）最有效的方法。MapReduce本身具有并行执行能力，并且可由你指定reduce的数量。 执行方式如下：
顺序文件(SequenceFile)的格式 顺序文件的格式是由文件头和随后的一条或多条记录组成。顺序文件的前三个字节为SEQ（顺序文件代码），紧随其后的是一个字节表示顺序文件的版本号。文件头还包括其他一些字段，包括键和值相应类的名称，数据压缩细节，用户定义的元数据，以及同步标识。同步标识主要用于读取文件的时候能够从任意位置开始识别记录便捷。每个文件有随机生成的同步标识，该标识内容存储在文件头中，同步标识位于顺序文件中的记录与记录之间。同步标识的额外存储开销要求小于1%，所以没有必要再每条记录末尾添加该标识。 记录的内部结构与是否启用压缩有关。如果启用，则与是记录压缩还是数据块压缩有关。如果没有启用压缩（默认情况），那么每条记录有记录长度（字节数），键长度，键和值组成。长度字段为4字节长的整数，并且需要遵循java.io.DataOutput类中writeInt()方法的协定。通过为数据写入顺序文件而定义的Serialization类，可以实现对键和值的序列化。 记录压缩的格式与无压缩情况相同，只不过值需要通过文件头中定义的压缩codec进行压缩。注意，键是不会压缩的
块压缩一次对多个记录进行压缩，因此相对于单条记录压缩，也所效率会更高，因为可以利用记录件的相似性进行压缩 如下图：
2).MapFile MapFile是已经排序的SequenceFile，它已加入用于搜索键的索引。可以将MapFile视为java.util.Map的持久化形式
A.写入MapFile MapFile写入类似于SequenceFile的写入
public class MapFileWriteDemo { private static final String[] DATA = { "One, two, buckle my shoe", "Three, four, shut the door", "Five, six, pick up sticks", "Seven, eight, lay them straight", "Nine, ten, a big fat hen" }; public static void main(String[] args) throws IOException {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
IntWritable key = new IntWritable();
Text value = new Text();
MapFile.Writer writer = null; try {
writer = new MapFile.Writer(conf, fs, uri,
key.getClass(), value.getClass()); for (int i = 0; i < 1024; i++) {
key.set(i + 1);
value.set(DATA[i % DATA.length]);
writer.append(key, value);
}
} finally {
IOUtils.closeStream(writer);
}
}
}
两个文件都是SequenceFile，data文件包含所有记录，依次为：
index文件包含一部分键和data文件中键偏移量的映射
在上面结果中，我们可以用MapFile.Write实例中的setIndexInterVal()方法来增加索引间隔数量，这样可以减少MapFile中用于存储索引的内存
B.读取MapFile 在MapFile文件依次遍历所有条目的过程类似SequenceFile中的过程：先建立MapFile.Reader实例，然后调用next()方法，知道返回值为false。通过调用get()方法可以随机访问文件中的数据。
C.将SequenceFile转换为MapFile 在MapFile中搜索就相当于在索引和排过序的SequenceFile中搜索，所以自然想到将SequenceFile转换为MapFile。前面已经介绍SequenceFile的排序，下面介绍MapFile调用fix()静态方法，该方法能够为MapFile重建索引
public class MapFileFixer { public static void main(String[] args) throws Exception {
String mapUri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(mapUri), conf);
Path map = new Path(mapUri);
Path mapData = new Path(map, MapFile.DATA_FILE_NAME); // Get key and value types from data sequence file SequenceFile.Reader reader = new SequenceFile.Reader(fs, mapData, conf);
Class keyClass = reader.getKeyClass();
Class valueClass = reader.getValueClass();
reader.close(); // Create the map file index file long entries = MapFile.fix(fs, map, keyClass, valueClass, false, conf);
System.out.printf("Created MapFile %s with %d entries\n", map, entries);
}
}
Fix()方法通常用于重建已损坏的做引，但是由于它是重头开始建立新的索引的 所以我们可以用它来为SequenceFile建立新的索引。方法见P127. (责任编辑：IT) |
