> Linux集群 > Hadoop >

Hadoop：The Definitive Guid 总结 Chapter 4 Hadoop I/O

时间:2014-12-30 23:41  来源:linux.it.net.cn   作者:IT

1.数据的完整性

1).HDFS的数据完整性

  • HDFS以透明方式校验所有写入它的数据，并在默认设置下，会在读取数据时验证校验和。针对数据的每个io.bytes.per.checksum字节都会创建一个单独的校验和。默认值为512字节；
  • DataNode负责在存储数据（包括数据的校验和）之前验证它们收到的数据，其中管道线的最后一个DataNode负责验证校验和，如果此datanode检测到错误，客户端会收到一个checksum Exception。
  • 客户端从datanode上读取数据时，也会验证校验和，将其与datanode上存储的校验和进行比较。每个datanode都维护着一个连续的校验和和验证日志，里面有着每个block的最后验证的时间。数据端成功验证block之后，便会告诉datanode，datanode随之更新日志。
  • 每个datanode会在后台线程运行一个DataBlockScanner定期验证存储在datanode上的所有block。这是为了防止物理存储介质出现衰减。

 

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();
    }
}
 


 使用ReflectionUtils创建一个codecs的实例。给createOutputStream传递一个标准输入进而获得一个封装了system.out的CompressionOutputStream，通过IOUtils的copyBytes方法把标准输入的数据拷贝到CompressionOutputStream中

 

通过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非常重要。

  • 对于gzip格式来说，因为gzip格式的stream不支持从任意点读取等原因，所以gzip并不支持文件切分，MapReduce也不会切分gzip压缩文件，牺牲了数据的本地性
  • Bzip2文件提供不同数据块之间的同步标识，因而它是支持切分的
  • Zip是一个归档文件，里面存放多个文件。文件的位置保存在zip文件尾的中央目录中，所以理论上它支持，但事实上Hadoop现在还没有能支持zip格式的split输入。

怎么挑选压缩格式，对于巨大的、没有存储边界的文件、如日志文件，可以考虑如下选项：

  • 存储未经压缩的文件
  • 使用支持切分的格式，像bzip2
  • 在程序中把文件分成chunks，对chunks进行单独压缩（使用任何压缩格式都行）。这种情况下，你需要选择chunk的大小，以期能与HDFS的block大小接近。
  • 使用Avro数据文件，该文件支持压缩和切分（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);
    }
}
 

 


如果输出的文件尾时和顺序文件(Sequence File)，可以设置mapred.output.compression.type属性来控制使用哪种压缩格式，默认值RECORD，针对条记录进行压缩，如果将其改为BLOCK，将针对一组记录进行压缩(推荐，因为压缩效率更高)也可以用SequenceFileOutputFormat的静态方法setOutputCompressionType()设置这个属性，下面给出MapReduce压缩属性：

 

实际上也可以对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));
 


 另外利用WritableComparable，RawComparator，WritableComparator等类 可以进行类型比较

 

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 

  • ObjectWritable是对java基本类型（string、enum、writable、null或这些类型组成的数组）的一个通用封装，它在Hadoop RPC中用于对方法的参数和返回类型进行封装和解封装
  • 作为一个通用机制，每次序列化都封装类型的名称，这非常浪费空间，如果封装的类型数量比较少并且能够提前知道，那么可以通过使用静态类型的数组，并使用对序列化后的类型的引用加入位置索引提高性能。这是 GenericWritable 类型采取的方法，并且你可以在继承子类中指定需要支持的类型 。

 

F.Writable集合类

org.apache.hadoop.io包中有四种Writable集合类型，分别是ArrayWritable，TwoDArrayWritable，MapperWritable和SortedMapWritable：

  • ArrayWritable和TwoDArrayWritable是Writable针对数组和二维数据（数组的数组）实例的实现。所有对ArrayWritable或者TwoDArrayWritable的使用都必须实例化相同的类，这是在构造时指定的；另外，ArrayWritable和TwoDArrayWritable都有get（）和set（）方法，也有toArray（）方法，后者用于创建数组（或者二维数组）的浅拷贝。
  • MapWritable和SortedMapWritable分别是java.util.Map(Writable,Writable)和java.util.SortedMap（WritableComparable，Writable）的实例。每个键/值字段的类型都是此字段序列化格式的一部分。

 

 

 

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) &amp;&amp; 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的数量。

执行方式如下：

 % hadoop jar $HADOOP_INSTALL/hadoop-*-examples.jar sort -r 1 \
-inFormat org.apache.hadoop.mapred.SequenceFileInputFormat \
-outFormat org.apache.hadoop.mapred.SequenceFileOutputFormat \
-outKey org.apache.hadoop.io.IntWritable \
-outValue org.apache.hadoop.io.Text \
numbers.seq sorted
% hadoop fs -text sorted/part-00000 | head

1     Nine, ten, a big fat hen
2     Seven, eight, lay them straight
3     Five, six, pick up sticks
4     Three, four, shut the door
5     One, two, buckle my shoe
6     Nine, ten, a big fat hen
7     Seven, eight, lay them straight
8     Five, six, pick up sticks
9     Three, four, shut the door
10    One, two, buckle my shoe

 

 

顺序文件(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);
    }
  }
}
 


如果我们观察MapFile，我们会发现它实际上是一个其中包含data和index两个文件的文件夹：

% ls -l numbers.map
total 104
-rw-r--r-- 1 tom tom 47898 Jul 29 22:06 data
-rw-r--r-- 1 tom tom 251 Jul 29 22:06 index

两个文件都是SequenceFile，data文件包含所有记录，依次为：

 % hadoop fs -text numbers.map/data | head
1                  One, two, buckle my shoe
2                  Three, four, shut the door
3                  Five, six, pick up sticks
4                  Seven, eight, lay them straight
5                  Nine, ten, a big fat hen
6                  One, two, buckle my shoe
7                  Three, four, shut the door
8                  Five, six, pick up sticks
9                  Seven, eight, lay them straight
10                Nine, ten, a big fat hen

index文件包含一部分键和data文件中键偏移量的映射

 % hadoop fs -text numbers.map/index
1                   128
129                6079
257                12054
385                18030
513                24002
641                29976
769                35947
897                41922

在上面结果中，我们可以用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.


转载地址：http://www.cnblogs.com/biyeymyhjob/archive/2012/08/10/2630484.html


(责任编辑：IT)
    最近关注
    热点内容