org.apache.hadoop.io.WritableUtils#writeVInt ( )源码实例Demo

private static Slice writeVintOld(SliceOutput output, long value)
        throws IOException
{
    output.reset();
    WritableUtils.writeVLong(output, value);
    Slice vLongOld = Slices.copyOf(output.slice());

    output.reset();
    RcFileDecoderUtils.writeVLong(output, value);
    Slice vLongNew = Slices.copyOf(output.slice());
    assertEquals(vLongNew, vLongOld);

    if (value == (int) value) {
        output.reset();
        WritableUtils.writeVInt(output, (int) value);
        Slice vIntOld = Slices.copyOf(output.slice());
        assertEquals(vIntOld, vLongOld);

        output.reset();
        RcFileDecoderUtils.writeVInt(output, (int) value);
        Slice vIntNew = Slices.copyOf(output.slice());
        assertEquals(vIntNew, vLongOld);
    }
    return vLongOld;
}
 

public void append(byte[] kvBuffer, int offset, int keyLength,
    int valueLength)
    throws IOException {
  int realKeyLen = keyLength + WritableUtils.INT_LENGTH_BYTES;
  int realValLen = valueLength + WritableUtils.INT_LENGTH_BYTES;

  WritableUtils.writeVInt(buffer, realKeyLen);
  WritableUtils.writeVInt(buffer, realValLen);
  //this is real key: keyLength + key
  buffer.writeInt(keyLength);
  buffer.write(kvBuffer, offset, keyLength);
  //this is real value: 
  buffer.writeInt(valueLength);
  buffer.write(kvBuffer, offset + keyLength, valueLength);

  out.write(buffer.getData(), 0, buffer.getLength());
  buffer.reset();

  // Update bytes written
  decompressedBytesWritten += realKeyLen + realValLen
      + WritableUtils.getVIntSize(realKeyLen)
      + WritableUtils.getVIntSize(realValLen);
  ++numRecordsWritten;
}
 

/**
 * Write an array of blocks as compactly as possible. This uses
 * delta-encoding for the generation stamp and size, following
 * the principle that genstamp increases relatively slowly,
 * and size is equal for all but the last block of a file.
 */
public static void writeCompactBlockArray(
    Block[] blocks, DataOutputStream out) throws IOException {
  WritableUtils.writeVInt(out, blocks.length);
  Block prev = null;
  for (Block b : blocks) {
    long szDelta = b.getNumBytes() -
        (prev != null ? prev.getNumBytes() : 0);
    long gsDelta = b.getGenerationStamp() -
        (prev != null ? prev.getGenerationStamp() : 0);
    out.writeLong(b.getBlockId()); // blockid is random
    WritableUtils.writeVLong(out, szDelta);
    WritableUtils.writeVLong(out, gsDelta);
    prev = b;
  }
}
 

/**
 * GenericGroup ::= displayName #counter counter*
 */
@Override
public synchronized void write(DataOutput out) throws IOException {
  Text.writeString(out, displayName);
  WritableUtils.writeVInt(out, counters.size());
  for(Counter counter: counters.values()) {
    counter.write(out);
  }
}
 

@Override
public void write(DataOutput out) throws IOException {
  Properties props = getProps();
  WritableUtils.writeVInt(out, props.size());
  for(Entry<Object, Object> item: props.entrySet()) {
    org.apache.hadoop.io.Text.writeString(out, (String) item.getKey());
    org.apache.hadoop.io.Text.writeString(out, (String) item.getValue());
    WritableUtils.writeCompressedStringArray(out,
        updatingResource.get(item.getKey()));
  }
}
 

@Override
public void write(DataOutput output) throws IOException {
    output.writeBoolean(isNullsLast);
    output.writeBoolean(isAscending);
    WritableUtils.writeVInt(output, ExpressionType.valueOf(expression).ordinal());
    expression.write(output);
}
 

public void close() throws IOException {
  // Write EOF_MARKER for key/value length
  WritableUtils.writeVInt(out, IFile.EOF_MARKER);
  WritableUtils.writeVInt(out, IFile.EOF_MARKER);
  
  // Close the stream 
  out.close();
  out = null;
}
 

@Override
public void write(DataOutput out, boolean reducedResponse) throws IOException {
    WritableUtils.writeVInt(out, _count);
    out.writeBoolean(trackSizes);
    // Write out the number of Attributes we're going to store
    WritableUtils.writeVInt(out, this.attributes.size());
    
    for (Attribute<? extends Comparable<?>> attr : this.attributes) {
        // Write out the concrete Attribute class
        WritableUtils.writeString(out, attr.getClass().getName());
        
        // Defer to the concrete instance to write() itself
        attr.write(out, reducedResponse);
    }
}
 

protected void writeValue(byte[] data, int offset, int length) throws IOException {
  writeRLE(out);
  WritableUtils.writeVInt(out, length); // value length
  out.write(data, offset, length);
  // Update bytes written
  decompressedBytesWritten +=
      length + WritableUtils.getVIntSize(length);
  if (serializedUncompressedBytes != null) {
    serializedUncompressedBytes.increment(length);
  }
  totalKeySaving++;
}
 

/** Append a key/value pair. */
@Override
@SuppressWarnings("unchecked")
public synchronized void append(Object key, Object val)
  throws IOException {
  if (key.getClass() != keyClass)
    throw new IOException("wrong key class: "+key+" is not "+keyClass);
  if (val.getClass() != valClass)
    throw new IOException("wrong value class: "+val+" is not "+valClass);

  // Save key/value into respective buffers 
  int oldKeyLength = keyBuffer.getLength();
  keySerializer.serialize(key);
  int keyLength = keyBuffer.getLength() - oldKeyLength;
  if (keyLength < 0)
    throw new IOException("negative length keys not allowed: " + key);
  WritableUtils.writeVInt(keyLenBuffer, keyLength);

  int oldValLength = valBuffer.getLength();
  uncompressedValSerializer.serialize(val);
  int valLength = valBuffer.getLength() - oldValLength;
  WritableUtils.writeVInt(valLenBuffer, valLength);
  
  // Added another key/value pair
  ++noBufferedRecords;
  
  // Compress and flush?
  int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();
  if (currentBlockSize >= compressionBlockSize) {
    sync();
  }
}
 

/**
 * GenericGroup ::= displayName #counter counter*
 */
@Override
public synchronized void write(DataOutput out) throws IOException {
  Text.writeString(out, displayName);
  WritableUtils.writeVInt(out, counters.size());
  for(TezCounter counter: counters.values()) {
    counter.write(out);
  }
}
 

/**
 */
@Override
public void write(DataOutput out) throws IOException {
  WritableUtils.writeVInt(out, keyId);
  WritableUtils.writeVLong(out, expiryDate);
  if (keyBytes == null) {
    WritableUtils.writeVInt(out, -1);
  } else {
    WritableUtils.writeVInt(out, keyBytes.length);
    out.write(keyBytes);
  }
}
 

@Override
public void write(DataOutput output) throws IOException {
    WritableUtils.writeVInt(output, (type.ordinal() + 1) * (this.isNullable ? -1 : 1));
    WritableUtils.writeVInt(output, count * (columnModifier == null ? 1 : -1));
    if (type.isFixedWidth() && type.getByteSize() == null) {
        WritableUtils.writeVInt(output, byteSize);
    }
}
 

@Override
public void write(DataOutput output) throws IOException {
    WritableUtils.writeVInt(output, ExpressionType.valueOf(children.get(0)).ordinal());
    children.get(0).write(output);
}
 

public void binaryProtocolStub() {
  try {

    initSoket();

    //should be 5
    //RUN_REDUCE boolean 
    WritableUtils.readVInt(dataInput);
    WritableUtils.readVInt(dataInput);
    int intValue = WritableUtils.readVInt(dataInput);
    System.out.println("getIsJavaRecordWriter:" + intValue);

    // reduce key
    WritableUtils.readVInt(dataInput);
    // value of reduce key
    BooleanWritable value = new BooleanWritable();
    readObject(value, dataInput);
    System.out.println("reducer key :" + value);
    // reduce value code:

    // reduce values
    while ((intValue = WritableUtils.readVInt(dataInput)) == 7) {
      Text txt = new Text();
      // value
      readObject(txt, dataInput);
      System.out.println("reduce value  :" + txt);
    }


    // done
    WritableUtils.writeVInt(dataOut, 54);

    dataOut.flush();
    dataOut.close();

  } catch (Exception x) {
    x.printStackTrace();
  } finally {
    closeSoket();

  }
}
 

public void write(DataOutput out) throws IOException {
    WritableUtils.writeVInt(out, name.length);
    for (int i = 0; i < name.length; i++) {
        datum[i].write(out);
    }
}
 

@Override
public void write(DataOutput output) throws IOException {
    WritableUtils.writeVInt(output, ExpressionType.valueOf(children.get(0)).ordinal());
    children.get(0).write(output);
}
 

public void write(DataOutput out) throws IOException {
  Text.writeString(out, mapId);
  WritableUtils.writeVLong(out, compressedLength);
  WritableUtils.writeVLong(out, uncompressedLength);
  WritableUtils.writeVInt(out, forReduce);
}
 

public void write(DataOutput out) throws IOException {
  WritableUtils.writeVInt(out, first);
  WritableUtils.writeVInt(out, length);
}
 

public void abort() throws IOException {
  WritableUtils.writeVInt(stream, MessageType.ABORT.code);
  LOG.debug("Sent abort command");
}