Spark 数据格式 UnsafeRow

case class Person(id: Long, id2: Long, id3: String)val e = Encoders.product[Person]val personExprEncoder = e.asInstanceOf[ExpressionEncoder[Person]]val person = Person(2, 7,"abcdefghijklmnopqrst")val row = personExprEncoder.toRow(person) //这是一个UnsafeRow对象，且baseObject为byte[64]，对于为什么为64，下文分析println(row.getLong(0))println(row.getString(2))

abstract class UnsafeProjection extends Projection {	override def apply(row: InternalRow): UnsafeRow}

class SpecificUnsafeProjection extends org.apache.spark.sql.catalyst.expressions.UnsafeProjection {
private Object[] references;private boolean resultIsNull_0;private boolean globalIsNull_0;private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[] mutableStateArray_2 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[1];private java.lang.String[] mutableStateArray_0 = new java.lang.String[1];private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] mutableStateArray_3 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[1];private UnsafeRow[] mutableStateArray_1 = new UnsafeRow[1]; 
public SpecificUnsafeProjection(Object[] references) {this.references = references;mutableStateArray_1[0] = new UnsafeRow(3); //创建UnsafeRow实例，3个字段：id,id2,id3mutableStateArray_2[0] = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(mutableStateArray_1[0], 32);mutableStateArray_3[0] = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(mutableStateArray_2[0], 3);
} public UnsafeRow apply(InternalRow i) {mutableStateArray_2[0].reset(); mutableStateArray_3[0].zeroOutNullBytes();writeFields_0_0(i); writeFields_0_1(i);mutableStateArray_1[0].setTotalSize(mutableStateArray_2[0].totalSize());return mutableStateArray_1[0];}//初始化字段id3private void writeFields_0_1(InternalRow i) {UTF8String value_13 = StaticInvoke_0(i);if (globalIsNull_0) {mutableStateArray_3[0].setNullAt(2);} else {mutableStateArray_3[0].write(2, value_13);}} //初始化字段id,id1private void writeFields_0_0(InternalRow i) {boolean isNull_3 = i.isNullAt(0);com.test.scala.EncoderScala$Person value_3 = isNull_3 ? null : ((com.test.scala.EncoderScala$Person) i.get(0, null));long value_0 = value_3.id();if (isNull_0) {mutableStateArray_3[0].setNullAt(0);} else {mutableStateArray_3[0].write(0, value_0);}com.test.scala.EncoderScala$Person value_7 = isNull_7 ? null : ((com.test.scala.EncoderScala$Person) i.get(0, null));long value_4 = value_7.id2();if (isNull_4) {mutableStateArray_3[0].setNullAt(1);} else {mutableStateArray_3[0].write(1, value_4);}}}

public BufferHolder(UnsafeRow row, int initialSize) {   int bitsetWidthInBytes = UnsafeRow.calculateBitSetWidthInBytes(row.numFields());  if (row.numFields() > (ARRAY_MAX - initialSize - bitsetWidthInBytes) / 8) {  throw new UnsupportedOperationException(  "Cannot create BufferHolder for input UnsafeRow because there are " +  "too many fields (number of fields: " + row.numFields() + ")");  }  this.fixedSize = bitsetWidthInBytes + 8 * row.numFields(); //固定长度   this.buffer = new byte[fixedSize + initialSize]; //即UnsafeRow.baseObject  this.row = row;  this.row.pointTo(buffer, buffer.length);}

初始内存 = fixedSize + initialSize          = (bitsetWidthInBytes + 8*总字段数) + (可变字段数*32)      = 8+8*3+1*32      = 64

public void write(int ordinal, UTF8String input) {  final int numBytes = input.numBytes(); //计算id3的字节数，20个字母，占20字节  final int roundedSize = ByteArrayMethods.roundNumberOfBytesToNearestWord(numBytes); //需要为8的位数，32>=20,32个字节都会分配给id3的值  holder.grow(roundedSize); //动态扩展内存，刚好initialSize为32，所以本次不需要扩展  zeroOutPaddingBytes(numBytes);  input.writeToMemory(holder.buffer, holder.cursor); //id3为第一个不可变字段，所以cursor刚好指向variable length portion区域的起始位置48  setOffsetAndSize(ordinal, numBytes); //设置id3的相对偏移量offset=(cursor-baseOffset)=32和size=numBytes=20  holder.cursor += roundedSize; //cursor往后移32字节，代表下一个不可变字段的offset}

@Overridepublic void writeExternal(ObjectOutput out) throws IOException {  byte[] bytes = getBytes();  out.writeInt(bytes.length);  out.writeInt(this.numFields);  out.write(bytes);}@Overridepublic void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {  this.baseOffset = BYTE_ARRAY_OFFSET;  this.sizeInBytes = in.readInt();  this.numFields = in.readInt();  this.bitSetWidthInBytes = calculateBitSetWidthInBytes(numFields);  this.baseObject = new byte[sizeInBytes];  in.readFully((byte[]) baseObject);}