Add record encoder

[harbby]

添加 数据行编码器/解码器功能

背景:

分布式计算系统中经常会通过shuffle在节点间传输大量数据．IO瓶颈(磁盘/网络)也shuffle环节最大的挑战．
通过减少传输字节数来提升IO是非常有效的手段．常见做法是高效的序列化器+压缩来提升IO性能．

特点:

• 高效的序列化器(编码器/解码器)+压缩可以显著减少字节数，大幅提升IO．
• 可用于字节存储．字节存储相比对象存储更加节省空间，且能显著提高Cpu Cache命中率．字节存储例子: Flink的BinaryRow，Spark的UnsafeRow等．

设计

所有编码器解码器都继承自Encoder<E>且由Encoders类进行引用:
接口设计如下:

public interface Encoder<E>
        extends Serializable
{
    public void encoder(E value, DataOutput output)
            throws IOException;

    public E decoder(DataInput input)
            throws IOException;
}

兼容性:

• 该patch不会引入破坏性Api变化．
• 阶段实验性加入 DataSet.encoder(Encoder) 方法.该方法会在类型推导功能(# ??? )完成后移除
• 该功能会对netty网络传输后端造成破坏性改变，需要在后端加入相应解码器

效果:

• 该功能将允许用户添加设置Record的序列化器(Encoder)．
• 该序列化器将在shuffleMap write和shufflerReduce reader时起作用，将显著降低传输的字节数．
• 且在小Record测试下有10倍数的提升(测试参考: ....)