记一次序列化导致的MR统计问题

作者: 天不沽 | 来源:发表于2018-01-17 16:40 被阅读62次

类似简单的wordcount，MapReduce输出的统计结果却不符预期。部分key丢失了，未丢失的key，其value值也不对。
经查，是因为代码中指定了SortComparator，导致key比较时所用的反序列化方式与序列化方式不符，这影响了map在merge时对key的排序，也影响了reduce对key的分组。

下面结合代码，对以下步骤进行梳理（代码来自hadoop-mapreduce-client-core-2.6.0-cdh5.5.0, hadoop-common-2.6.0-cdh5.5.0, avro-1.7.6）。

map端key的序列化器；
map端merge时key的比较器；
reduce端对key的分组；
reduce端key分组时的比较器；
key的比较。

map端key的序列化器——keySerializer

Map的输出主要由org.apache.hadoop.mapred.MapTask的内部类MapOutputBuffer实现，MapOutputBuffer使用keySerializer对key进行序列化。

keySerializer是什么呢？
keySerializer本身是个接口，具体实现类是在MapOutPutBuffer的init()方法中指定的，代码如下：

public void init(Context context) throws ... {
  ...
  this.keySerializer = this.serializationFactory.getSerializer(this.keyClass);
  ...
}

SerializationFactory又是如何根据keyClass得到对应的Serializer的呢？
SerializationFactory维护了一个链表serializations，SerializationFactory会依次遍历serializations中的serialization，找到第一个accept keyclass的serialization。代码如下：

public <T> Serialization<T> getSerialization(Class<T> c) {
    Iterator i$ = this.serializations.iterator();

    Serialization serialization;
    do {
      if(!i$.hasNext()) {
        return null;
      }

      serialization = (Serialization)i$.next();
    } while(!serialization.accept(c)); //是否accept class

    return serialization;
  }

serializations又有哪些呢？
SerializationFactory的构造函数会读取io.serializations配置来初始化serializations。若没有配置io.serializations，则取默认的serializations，依次为WritableSerialization、AvroSpecificSerialization和AvroReflectSerialization。代码如下：

private List<Serialization<?>> serializations = new ArrayList();

public SerializationFactory(Configuration conf) {
  ...
  String[] arr$ = conf.getStrings("io.serializations", new String[]{WritableSerialization.class.getName(), AvroSpecificSerialization.class.getName(), AvroReflectSerialization.class.getName()});
  ...
}

一个serialization是否accept某个class，则是由具体的serialization自己实现的，如WritableSerialization，其接受的是Writable的子类。WritableSerialization的accept()代码如下：

public boolean accept(Class<?> c) {
    return Writable.class.isAssignableFrom(c);
}

map端merge时key的比较器

MapOutputBuffer在对数据进行merge时，使用了比较器对key进行排序。

Merger.merge(this.job, this.rfs, this.keyClass, this.valClass, this.codec, segmentList, this.job.getInt("io.sort.factor", 100), new Path(mapId.toString()), 
this.job.getOutputKeyComparator(),  //获取key的比较器
this.reporter, (Counter)null, this.mapTask.spilledRecordsCounter);

getOutputKeyComparator()获取的又是什么呢？获取的是mapreduce.job.output.key.comparator.class配置的比较器。

reduce端对key的分组

reduce端按key分组处理的主流程在org.apache.hadoop.mapreduce.Reducer的run()方法中（这个Reducer就是我们在写MR应用程序reduce的部分时，继承的Reducer），run()方法的代码如下：

public void run(Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws ... {
    this.setup(context);

    try {
      while(context.nextKey()) {
        this.reduce(context.getCurrentKey(), context.getValues(), context);
        ...
      }
    } finally {
      this.cleanup(context);
    }

  }

这个run()方法里的setup(), reduce()和cleanup()就是经常被我们重写的三个方法。

key分组的关键，在于理解context的具体实现——ReduceContextImpl。

ReduceContextImpl

ReduceContextImpl的关键成员如下（其中KEYIN和VALUEIN都是泛型）。

public class ReduceContextImpl ... {
  private KEYIN key;
  private VALUEIN value;
  private boolean nextKeyIsSame = false;
  public boolean nextKey();
  public boolean nextKeyValue();
  protected class ValueIterator ... {
   public boolean hasNext();
   public VALUEIN next();
  }
}

其中，四个主要方法的含义如下

nextKeyValue()：读取下对(key, value)
nextKey()：读取下一个key，其实是调用了nextKeyValue()，同时也读取了下一个value（其实，map写过来的就是一对对(key, value)，读的话，也应该是一对对的读）
values.hasNext()：即ValueIterator.hasNext()，判断是否还有下一个value
values.next()：即ValueIterator.next()，读取下一个value，其实也是调用了nextKeyValue()，同时也读取了下一个key

nextKeyValue()每次读取下一对(key, value)，都将key值和value值存储在KEYIN key和VALUEIN value中（这就是reduce端，key/value的复用）。

这里主要阐述ReduceContextImpl 是如何实现对key的分组处理。为了说明方便，这里将上文中Reducer.run()方法展开如下（其中，假设用户实现的reduce()使用了iterator的形式对value进行了遍历）。

public void run(Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context context) throws ... {
  while(context.nextKey()) { //外部while
    while(values.hasNext()) { // 内部while，即用户重写的reduce()方法
      context.write(key, values.next());
    }
    ...
  }
}

key的分组处理，关键在于nextKeyValue()方法。上述代码中，未直接体现nextKeyValue()方法，而我们知道values.next()调用了该方法。nextKeyValue()方法在读取到当前key后，会将当前key与下一个将要处理的key进行比较，若发现不是同一个key，就会将nextKeyIsSame置为false。此后，values.hasNext()方法发现nextKeyIsSame为false，也会返回false，从而跳出循环，结束一个key的处理。这样，就保证了每一次内部while，处理的都是同一个key。
下面，将几处关键代码摘要如下（这里面省略了firstValue以及很多其他的处理逻辑）。

public boolean nextKeyValue() {
  DataInputBuffer nextKey = this.input.getKey(); //获取当前key
  this.currentRawKey.set(nextKey.getData(), nextKey.getPosition(), nextKey.getLength() - nextKey.getPosition());
  this.key = this.keyDeserializer.deserialize(this.key); //反序列化当前key
  
  DataInputBuffer nextVal = this.input.getValue(); //获取当前value
  this.value = this.valueDeserializer.deserialize(this.value); //反序列化当前value
  
  this.hasMore = this.input.next(); //是否还有下一个(key, value)
  if(this.hasMore) {
    //如果有下一个(key,, value)，获取下一个key
    nextKey = this.input.getKey();
    //比较当前key和下一个key，设置nextKeyIsSame
    this.nextKeyIsSame = this.comparator.compare(this.currentRawKey.getBytes(), 0, this.currentRawKey.getLength(), nextKey.getData(), nextKey.getPosition(), nextKey.getLength() - nextKey.getPosition()) == 0;
  } else {
    this.nextKeyIsSame = false;
  }
}

public boolean hasNext() {
  ...
  return ReduceContextImpl.this.nextKeyIsSame;
}

reduce端key分组时的比较器

我们注意到，nextKeyValue()在比较当前key和下一个key时，用到了一个比较器。这个比较器是在org.apache.hadoop.mapred.ReduceTask.run()方法中获取的。

public void run(JobConf job, TaskUmbilicalProtocol umbilical) throws ... {
      ...
      RawComparator comparator = job.getOutputValueGroupingComparator(); // 获取比较器
      if(useNewApi) {
        this.runNewReducer(job, umbilical, reporter, rIter, comparator, keyClass, valueClass);
      } else {
        ...
      }
      ...
}

getOutputValueGroupingComparator()按如下优先级获取比较器：

首先，获取mapreduce.job.output.group.comparator.class配置的分组比较器；
其次，获取mapreduce.job.output.key.comparator.class配置的key比较器（这里就和上文的“map端merge时key的比较器”相同了）；
最后，获取如Text、IntWritable这些类型自己定义的比较器。

这里再对第三点做点补充说明，像Text、IntWritable这些类型，都有一段static代码，将自己定义的Comparator注册到WritableComparator中。

static {
  // register this comparator
  WritableComparator.define(Text.class, new Comparator());
}

key的比较

在对key进行比较时，是根据序列化后的字节码进行比较的。comparator.compare()的形式如下：

public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2);

假设key的类型是Text，获取到的比较器是Text定义的Comparator。Text在比较时，会先读取一个整数，这个整数存储了Text的长度，再根据这个长度比较Text的内容（也就是说Text在存储的时候，先是存储了Text的长度，再存储Text内容本身）。

public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
  int n1 = WritableUtils.decodeVIntSize(b1[s1]);
  int n2 = WritableUtils.decodeVIntSize(b2[s2]);
  return compareBytes(b1, s1+n1, l1-n1, b2, s2+n2, l2-n2);
}

可以看到，在对Text类型的key进行比较时，没有将整个Text都反序列化，而时尽量在字节码的层面上进行比较。

经过上面的梳理，我们知道了

key在map端输出时，用keySeriailzer进行了序列化；
key在map端merge时，用comparator在序列化后的结果上进行比较；
key在reduce端分组时，用comparator在序列化后的结果上进行比较；
comparator在比较的时候，因为基于的是序列化后的结果，所以涉及到一些反序列化。

下面，我们来详细阐述下本文开头的问题。

问题是如何发生的？

问题代码中

配置了用户自定义的SortComparator（也就是配置了mapred.output.key.comparator.class）。这个用户定义的比较器在比较时，用的是org.apache.avro.io.BinaryData的compare，而BinaryData在比较时，序列化和反序列化用的是BinaryEncoder和BinaryDecoder。
key是Text类型，没有对keySerializer进行特殊配置，所以获取到的是默认的WritableSerialization，最终调用的是Text的序列化方法。

这导致key使用Text进行了序列化，但在比较时，却尝试使用BinaryDecoder进行反序列化。序列化和反序列化不符，导致map端的merge和reduce端的key分组，在进行比较时都出错了，所以最后统计结果不对。

最后再补充一些细节。

BinaryData的比较方法

BinaryData是根据BinaryDecoder（可以读取key的值）和key的schma进行比较的。这里的schema在avro中用来标识类型，如RECORD可理解为复杂的Class类型，而SRING、LONG等的含义显而易见。
下面的代码摘取了对RECORD和STRING类型进行比较时的一些细节（省略了一些和avro Field处理有关的细节）。我们可以看到，在对RECORD进行处理时，使用了递归的方式，并且在遇到第一个不相等的字段时，便返回结果，不再处理后续的字节码，也就节省了部分反序列化的时间。而对STRING的处理，和Text很相似，都是先读取文本的长度，再根据该长度比较文本本身。

private static int compare(Decoders d, Schema schema) throws IOException {
    Decoder d1 = d.d1; Decoder d2 = d.d2;
    switch (schema.getType()) {
    case RECORD: {
      for (Field field : schema.getFields()) {
        int c = compare(d, field.schema()); //递归调用
        if (c != 0) //一旦遇到一个字段不相等，直接返回
          return (field.order() != Field.Order.DESCENDING) ? c : -c;
      }
      return 0;
    }
  ...
    case STRING: {
      int l1 = d1.readInt();
      int l2 = d2.readInt();
      int c = compareBytes(d.d1.getBuf(), d.d1.getPos(), l1,
                           d.d2.getBuf(), d.d2.getPos(), l2);
      d.d1.skipFixed(l1);
      d.d2.skipFixed(l2);
      return c;
    }
...

BinaryData的序列化方式

可见avro官网的Binary Encoding一节。

记一次序列化导致的MR统计问题

map端key的序列化器——keySerializer

map端merge时key的比较器

reduce端对key的分组

ReduceContextImpl

reduce端key分组时的比较器

key的比较

问题是如何发生的？

BinaryData的比较方法

BinaryData的序列化方式

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读