Writable、WritableComparable和comparators

时间：2021-09-03 16:43:15

hadoop的序列化格式

hadoop自身的序列化存储格式就是实现了Writable接口的类,他只实现了前面两点，压缩和快速。但是不容易扩展，也不跨语言。
我们先来看下Writable接口，Writable接口定义了两个方法：
1.将数据写入到二进制流中
2.从二进制数据流中读取数据
  1. package org.apache.hadoop.io;
  2. public interface Writable {
  3. void write(java.io.DataOutput p1) throws java.io.IOException;
  4. void readFields(java.io.DataInput p1) throws java.io.IOException;
  5. }
我们再来看下Writable接口与序列化和反序列化是如何关联的：
  1. package com.sweetop.styhadoop;
  2. import junit.framework.Assert;
  3. import org.apache.hadoop.io.IntWritable;
  4. import org.apache.hadoop.io.Writable;
  5. import org.apache.hadoop.util.StringUtils;
  6. import org.junit.Before;
  7. import org.junit.Test;
  8. import java.io.*;
  9. /**
  10. * Created with IntelliJ IDEA.
  11. * User: lastsweetop
  12. * Date: 13-7-4
  13. * Time: 下午10:25
  14. * To change this template use File | Settings | File Templates.
  15. */
  16. public class TestWritable {
  17. byte[] bytes=null;
  18. /**
  19. * 初始化一个IntWritable实例，并且调用系列化方法
  20. * @throws IOException
  21. */
  22. @Before
  23. public void init() throws IOException {
  24. IntWritable writable = new IntWritable(163);
  25. bytes = serialize(writable);
  26. }
  27. /**
  28. * 一个IntWritable序列号后的四个字节的字节流
  29. * 并且使用big-endian的队列排列
  30. * @throws IOException
  31. */
  32. @Test
  33. public void testSerialize() throws IOException {
  34. Assert.assertEquals(bytes.length,4);
  35. Assert.assertEquals(StringUtils.byteToHexString(bytes),"000000a3");
  36. }
  37. /**
  38. * 创建一个没有值的IntWritable对象，并且通过调用反序列化方法将bytes的数据读入到它里面
  39. * 通过调用它的get方法，获得原始的值，163
  40. */
  41. @Test
  42. public void testDeserialize() throws IOException {
  43. IntWritable newWritable = new IntWritable();
  44. deserialize(newWritable,bytes);
  45. Assert.assertEquals(newWritable.get(),163);
  46. }
  47. /**
  48. * 将一个实现了Writable接口的对象序列化成字节流
  49. * @param writable
  50. * @return
  51. * @throws IOException
  52. */
  53. public static byte[] serialize(Writable writable) throws IOException {
  54. ByteArrayOutputStream out = new ByteArrayOutputStream();
  55. DataOutputStream dataOut = new DataOutputStream(out);
  56. writable.write(dataOut);
  57. dataOut.close();
  58. return out.toByteArray();
  59. }
  60. /**
  61. * 将字节流转化为实现了Writable接口的对象
  62. * @param writable
  63. * @param bytes
  64. * @return
  65. * @throws IOException
  66. */
  67. public static byte[] deserialize(Writable writable,byte[] bytes) throws IOException {
  68. ByteArrayInputStream in=new ByteArrayInputStream(bytes);
  69. DataInputStream dataIn = new DataInputStream(in);
  70. writable.readFields(dataIn);
  71. dataIn.close();
  72. return bytes;
  73. }
  74. }

WritableComparable和comparators

IntWritable实现了WritableComparable，接口看下源代码知道，WritableComparable是Writable接口和java.lang.Comparable<T>的一个子接口。
  1. package org.apache.hadoop.io;
  2. public interface WritableComparable <T>  extends org.apache.hadoop.io.Writable, java.lang.Comparable<T> {
  3. }

MapReduce在排序部分要根据key值的大小进行排序，因此类型的比较相当重要，RawComparator是Comparator的增强版

  1. package org.apache.hadoop.io;
  2. public interface RawComparator <T>  extends java.util.Comparator<T> {
  3. int compare(byte[] bytes, int i, int i1, byte[] bytes1, int i2, int i3);
  4. }

它可以做到，不先反序列化就可以直接比较二进制字节流的大小:

  1. package com.sweetop.styhadoop;
  2. import org.apache.hadoop.io.IntWritable;
  3. import org.apache.hadoop.io.RawComparator;
  4. import org.apache.hadoop.io.Writable;
  5. import org.apache.hadoop.io.WritableComparator;
  6. import org.eclipse.jdt.internal.core.Assert;
  7. import org.junit.Before;
  8. import org.junit.Test;
  9. import java.io.ByteArrayOutputStream;
  10. import java.io.DataOutputStream;
  11. import java.io.IOException;
  12. /**
  13. * Created with IntelliJ IDEA.
  14. * User: lastsweetop
  15. * Date: 13-7-5
  16. * Time: 上午1:26
  17. * To change this template use File | Settings | File Templates.
  18. */
  19. public class TestComparator {
  20. RawComparator<IntWritable> comparator;
  21. IntWritable w1;
  22. IntWritable w2;
  23. /**
  24. * 获得IntWritable的comparator,并初始化两个IntWritable
  25. */
  26. @Before
  27. public void init() {
  28. comparator = WritableComparator.get(IntWritable.class);
  29. w1 = new IntWritable(163);
  30. w2 = new IntWritable(76);
  31. }
  32. /**
  33. * 比较两个对象大小
  34. */
  35. @Test
  36. public void testComparator() {
  37. Assert.isTrue(comparator.compare(w1, w2) > 0);
  38. }
  39. /**
  40. * 序列号后进行直接比较
  41. * @throws IOException
  42. */
  43. @Test
  44. public void testcompare() throws IOException {
  45. byte[] b1 = serialize(w1);
  46. byte[] b2 = serialize(w2);
  47. Assert.isTrue(comparator.compare(b1, 0, b1.length, b2, 0, b2.length) > 0);
  48. }
  49. /**
  50. * 将一个实现了Writable接口的对象序列化成字节流
  51. *
  52. * @param writable
  53. * @return
  54. * @throws java.io.IOException
  55. */
  56. public static byte[] serialize(Writable writable) throws IOException {
  57. ByteArrayOutputStream out = new ByteArrayOutputStream();
  58. DataOutputStream dataOut = new DataOutputStream(out);
  59. writable.write(dataOut);
  60. dataOut.close();
  61. return out.toByteArray();
  62. }
  63. }

相关文章