java类org.apache.hadoop.io.VLongWritable的实例源码

@Override
public void init() throws IOException {
  registerKey(NullWritable.class.getName(), NullWritableSerializer.class);
  registerKey(Text.class.getName(), TextSerializer.class);
  registerKey(LongWritable.class.getName(), LongWritableSerializer.class);
  registerKey(IntWritable.class.getName(), IntWritableSerializer.class);
  registerKey(Writable.class.getName(), DefaultSerializer.class);
  registerKey(BytesWritable.class.getName(), BytesWritableSerializer.class);
  registerKey(BooleanWritable.class.getName(), BoolWritableSerializer.class);
  registerKey(ByteWritable.class.getName(), ByteWritableSerializer.class);
  registerKey(FloatWritable.class.getName(), FloatWritableSerializer.class);
  registerKey(DoubleWritable.class.getName(), DoubleWritableSerializer.class);
  registerKey(VIntWritable.class.getName(), VIntWritableSerializer.class);
  registerKey(VLongWritable.class.getName(), VLongWritableSerializer.class);

  LOG.info("Hadoop platform inited");
}

@Override
public void init() throws IOException {
  registerKey(NullWritable.class.getName(), NullWritableSerializer.class);
  registerKey(Text.class.getName(), TextSerializer.class);
  registerKey(LongWritable.class.getName(), LongWritableSerializer.class);
  registerKey(IntWritable.class.getName(), IntWritableSerializer.class);
  registerKey(Writable.class.getName(), DefaultSerializer.class);
  registerKey(BytesWritable.class.getName(), BytesWritableSerializer.class);
  registerKey(BooleanWritable.class.getName(), BoolWritableSerializer.class);
  registerKey(ByteWritable.class.getName(), ByteWritableSerializer.class);
  registerKey(FloatWritable.class.getName(), FloatWritableSerializer.class);
  registerKey(DoubleWritable.class.getName(), DoubleWritableSerializer.class);
  registerKey(VIntWritable.class.getName(), VIntWritableSerializer.class);
  registerKey(VLongWritable.class.getName(), VLongWritableSerializer.class);

  LOG.info("Hadoop platform inited");
}

@Override
public void compute(Iterable<DoubleWritable> messages) throws IOException {
    if (getSuperstep() == 0) {
        setValue(new DoubleWritable(Double.MAX_VALUE));
    }

    double minDist = isSource() ? 0d : Double.MAX_VALUE;
    for (DoubleWritable message : messages) {
        minDist = Math.min(minDist, message.get());
    }

    if (minDist < getValue().get()) {
        setValue(new DoubleWritable(minDist));
        for (Edge<VLongWritable, DoubleWritable> edge : getEdges()) {
            double distance = minDist + edge.getValue().get();
            sendMessage(edge.getTargetVertexId(), new DoubleWritable(distance));
        }
    }

    voteToHalt();
}

private Job buildJob2(Path input, Path output) throws Exception {
    Configuration conf = getConf();
    conf.setInt(Constants.PROP_BLOCK_SIZE, blockSize);

    Job job = new Job(conf, "data-piqid.piqconnect.IterationStage2");
    job.setJarByClass(Runner.class);

    job.setMapperClass(Mapper.class);
    job.setReducerClass(IterationStage2._Reducer.class);
    job.setNumReduceTasks(numberOfReducers);
    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);
    job.setMapOutputKeyClass(VLongWritable.class);
    job.setMapOutputValueClass(BlockWritable.class);
    job.setOutputKeyClass(BlockIndexWritable.class);
    job.setOutputValueClass(BlockWritable.class);
    job.setSortComparatorClass(VLongWritableComparator.class);

    SequenceFileInputFormat.setInputPaths(job, input);
    FileOutputFormat.setOutputPath(job, output);
    FileOutputFormat.setCompressOutput(job, true);

    setCompression(job);
    return job;
}

private Job buildJob3(Path input, Path output) throws Exception {
    Configuration conf = getConf();
    conf.setInt(Constants.PROP_BLOCK_SIZE, blockSize);

    Job job = new Job(conf, "data-piqid.piqconnect.FinalResultBuilder");
    job.setJarByClass(Runner.class);

    job.setMapperClass(FinalResultBuilder._Mapper.class);
    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setNumReduceTasks(0);
    job.setOutputKeyClass(VLongWritable.class);
    job.setOutputValueClass(VLongWritable.class);

    FileInputFormat.setInputPaths(job, input);
    FileOutputFormat.setOutputPath(job, output);
    FileOutputFormat.setCompressOutput(job, true);

    setCompression(job);
    return job;
}

@Override
public void reduce(VLongWritable key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException {
    long     start_node;
    long     end_node;
    String[] line;

    for (Text value : values) {
        line = value.toString().split("\t");
        start_node = Long.parseLong(line[0]);
        end_node = Long.parseLong(line[1]);
        for (long i = start_node; i <= end_node; i++) {
            KEY.set(i);
            VALUE.set(Long.toString(i));
            ctx.write(KEY, VALUE);
        }
    }
}

private Job buildJob() throws Exception {
    Configuration conf = getConf();
    conf.setLong("numberOfNodes", numberOfNodes);

    Job job = new Job(conf, "data-piqid.piqconnect.ConCmptIVGen_Stage1");
    job.setJarByClass(InitialVectorGenerator.class);
    job.setMapperClass(_Mapper.class);
    job.setReducerClass(_Reducer.class);
    job.setNumReduceTasks(numberOfReducers);
    job.setOutputKeyClass(VLongWritable.class);
    job.setOutputValueClass(Text.class);

    FileInputFormat.setInputPaths(job, pathBitmask);
    FileOutputFormat.setOutputPath(job, pathVector);
    FileOutputFormat.setCompressOutput(job, true);

    return job;
}

@Test
public void reduce1() throws IOException {
    reduceDriver.getConfiguration().setInt(Constants.PROP_BLOCK_SIZE, 2);

    reduceDriver.addInput(new VLongWritable(0), Arrays.asList(
            blockVector(TYPE.VECTOR_INITIAL, 0, 1),
            blockVector(TYPE.VECTOR_INCOMPLETE, 0, 0),
            blockVector(TYPE.VECTOR_INCOMPLETE, 2, 2)));

    reduceDriver.addInput(new VLongWritable(1), Arrays.asList(
            blockVector(TYPE.VECTOR_INITIAL, 2, 3),
            blockVector(TYPE.VECTOR_INCOMPLETE, 0, 1),
            blockVector(TYPE.VECTOR_INCOMPLETE, 2, 3)));

    reduceDriver.addOutput(blockIndex(0), blockVector(TYPE.VECTOR_INCOMPLETE, 0, 0));
    reduceDriver.addOutput(blockIndex(1), blockVector(TYPE.VECTOR_INCOMPLETE, 0, 1));

    reduceDriver.runTest();
}

@Test
public void reduce() throws IOException {
    reduceDriver.getConfiguration().setInt(Constants.PROP_BLOCK_SIZE, 3);

    int block_col = 0;

    BlockWritable e1 = blockVector(0, 1, 2);
    BlockWritable e2 = blockMatrix(block_col, 0, 1, 1, 0, 1, 2, 2, 1);

    reduceDriver.addInput(new IterationStage1.JoinKey(true, block_col), Arrays.asList(e1, e2));

    BlockWritable v1 = blockVector(BlockWritable.TYPE.VECTOR_INITIAL, 0, 1, 2);
    BlockWritable v2 = blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, 1, 0, 1);

    reduceDriver.addOutput(new VLongWritable(block_col), v1); // initial vector
    reduceDriver.addOutput(new VLongWritable(block_col), v2); // after multiplication
    reduceDriver.runTest();
}

@Test
public void reduce2() throws IOException {
    reduceDriver.getConfiguration().setInt(Constants.PROP_BLOCK_SIZE, 3);

    int block_col = 0;

    BlockWritable e1 = blockVector(1, -1, -1);
    BlockWritable e2 = blockMatrix(block_col, 2, 0);

    reduceDriver.addInput(new IterationStage1.JoinKey(true, block_col), Arrays.asList(e1, e2));

    BlockWritable v1 = blockVector(BlockWritable.TYPE.VECTOR_INITIAL, 1, -1, -1);
    BlockWritable v2 = blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, -1, -1, 1);

    reduceDriver.addOutput(new VLongWritable(block_col), v1); // initial vector
    reduceDriver.addOutput(new VLongWritable(block_col), v2); // after multiplication
    reduceDriver.runTest();
}

@Test
public void mapReduce2() throws IOException {
    mrDriver.getConfiguration().setInt(Constants.PROP_BLOCK_SIZE, 2);

    mrDriver.addInput(blockIndex(0), blockVector(0, 1));
    mrDriver.addInput(blockIndex(1), blockVector(2, 3));

    mrDriver.addInput(blockIndex(0, 0), blockMatrix(0L, 0, 1, 1, 0));
    mrDriver.addInput(blockIndex(0, 1), blockMatrix(1L, 1, 0));
    mrDriver.addInput(blockIndex(1, 0), blockMatrix(0L, 0, 1));
    mrDriver.addInput(blockIndex(1, 1), blockMatrix(1L, 1, 1));

    mrDriver.addOutput(new VLongWritable(0), blockVector(BlockWritable.TYPE.VECTOR_INITIAL, 0, 1));
    mrDriver.addOutput(new VLongWritable(0), blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, 1, 0));
    mrDriver.addOutput(new VLongWritable(1), blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, 1, -1));
    mrDriver.addOutput(new VLongWritable(1), blockVector(BlockWritable.TYPE.VECTOR_INITIAL, 2, 3));
    mrDriver.addOutput(new VLongWritable(0), blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, -1, 2));
    mrDriver.addOutput(new VLongWritable(1), blockVector(BlockWritable.TYPE.VECTOR_INCOMPLETE, -1, 3));

    mrDriver.runTest();
}

public static Value getValue(final VLongWritable w) {
  if (w == null) {
    throw new IllegalArgumentException("Writable cannot be null");
  }
  ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
  DataOutputStream out = new DataOutputStream(byteStream);

  // We could also close it, but we know that VLongWritable and BAOS don't need it.
  try {
    w.write(out);
  } catch (IOException e) {
    // If this ever happens, some seriously screwed up is happening or someone subclasses VLongWritable
    // and made it do crazy stuff.
    throw new RuntimeException(e);
  }

  return new Value(byteStream.toByteArray());
}

public static VLongWritable getWritable(final Value v) {
  if (null == v) {
    throw new IllegalArgumentException("Value cannot be null");
  }

  ByteArrayInputStream bais = new ByteArrayInputStream(v.get());
  DataInputStream in = new DataInputStream(bais);

  VLongWritable writable = new VLongWritable();
  try {
    writable.readFields(in); 
  } catch (IOException e) {
    // If this ever happens, some seriously screwed up is happening or someone subclasses Value
    // and made it do crazy stuff.
    throw new RuntimeException(e);
  }

  return writable;
}

public void write(Writable w) throws IOException {
  if (w instanceof TypedBytesWritable) {
    writeTypedBytes((TypedBytesWritable) w);
  } else if (w instanceof BytesWritable) {
    writeBytes((BytesWritable) w);
  } else if (w instanceof ByteWritable) {
    writeByte((ByteWritable) w);
  } else if (w instanceof BooleanWritable) {
    writeBoolean((BooleanWritable) w);
  } else if (w instanceof IntWritable) {
    writeInt((IntWritable) w);
  } else if (w instanceof VIntWritable) {
    writeVInt((VIntWritable) w);
  } else if (w instanceof LongWritable) {
    writeLong((LongWritable) w);
  } else if (w instanceof VLongWritable) {
    writeVLong((VLongWritable) w);
  } else if (w instanceof FloatWritable) {
    writeFloat((FloatWritable) w);
  } else if (w instanceof DoubleWritable) {
    writeDouble((DoubleWritable) w);
  } else if (w instanceof Text) {
    writeText((Text) w);
  } else if (w instanceof ArrayWritable) {
    writeArray((ArrayWritable) w);
  } else if (w instanceof MapWritable) {
    writeMap((MapWritable) w);
  } else if (w instanceof SortedMapWritable) {
    writeSortedMap((SortedMapWritable) w);
  } else if (w instanceof Record) {
    writeRecord((Record) w);
  } else {
    writeWritable(w); // last resort
  }
}

public Class<? extends Writable> readType() throws IOException {
  Type type = in.readType();
  if (type == null) {
    return null;
  }
  switch (type) {
  case BYTES:
    return BytesWritable.class;
  case BYTE:
    return ByteWritable.class;
  case BOOL:
    return BooleanWritable.class;
  case INT:
    return VIntWritable.class;
  case LONG:
    return VLongWritable.class;
  case FLOAT:
    return FloatWritable.class;
  case DOUBLE:
    return DoubleWritable.class;
  case STRING:
    return Text.class;
  case VECTOR:
    return ArrayWritable.class;
  case MAP:
    return MapWritable.class;
  case WRITABLE:
    return Writable.class;
  default:
    throw new RuntimeException("unknown type");
  }
}

public static void updateObject(Writable obj, byte[] seed) {
  if (obj instanceof IntWritable) {
    ((IntWritable)obj).set(Ints.fromByteArray(seed));
  } else if (obj instanceof FloatWritable) {
    ((FloatWritable)obj).set(r.nextFloat());
  } else if (obj instanceof DoubleWritable) {
    ((DoubleWritable)obj).set(r.nextDouble());
  } else if (obj instanceof LongWritable) {
    ((LongWritable)obj).set(Longs.fromByteArray(seed));
  } else if (obj instanceof VIntWritable) {
    ((VIntWritable)obj).set(Ints.fromByteArray(seed));
  } else if (obj instanceof VLongWritable) {
    ((VLongWritable)obj).set(Longs.fromByteArray(seed));
  } else if (obj instanceof BooleanWritable) {
    ((BooleanWritable)obj).set(seed[0] % 2 == 1 ? true : false);
  } else if (obj instanceof Text) {
    ((Text)obj).set(BytesUtil.toStringBinary(seed));
  } else if (obj instanceof ByteWritable) {
    ((ByteWritable)obj).set(seed.length > 0 ? seed[0] : 0);
  } else if (obj instanceof BytesWritable) {
    ((BytesWritable)obj).set(seed, 0, seed.length);
  } else if (obj instanceof UTF8) {
    ((UTF8)obj).set(BytesUtil.toStringBinary(seed));
  } else if (obj instanceof MockValueClass) {
    ((MockValueClass)obj).set(seed);
  } else {
    throw new IllegalArgumentException("unknown writable: " +
                                       obj.getClass().getName());
  }
}

public static <VTYPE> byte[] toBytes(VTYPE obj) {
  final String className = obj.getClass().getName();
  if (className.equals(IntWritable.class.getName())) {
    return Ints.toByteArray(((IntWritable) obj).get());
  } else if (className.equals(FloatWritable.class.getName())) {
    return BytesUtil.toBytes(((FloatWritable) obj).get());
  } else if (className.equals(DoubleWritable.class.getName())) {
    return BytesUtil.toBytes(((DoubleWritable) obj).get());
  } else if (className.equals(LongWritable.class.getName())) {
    return Longs.toByteArray(((LongWritable) obj).get());
  } else if (className.equals(VIntWritable.class.getName())) {
    return Ints.toByteArray(((VIntWritable) obj).get());
  } else if (className.equals(VLongWritable.class.getName())) {
    return Longs.toByteArray(((VLongWritable) obj).get());
  } else if (className.equals(BooleanWritable.class.getName())) {
    return BytesUtil.toBytes(((BooleanWritable) obj).get());
  } else if (className.equals(Text.class.getName())) {
    return ((Text)obj).copyBytes();
  } else if (className.equals(ByteWritable.class.getName())) {
    return Ints.toByteArray((int) ((ByteWritable) obj).get());
  } else if (className.equals(BytesWritable.class.getName())) {
    // TODO: copyBytes instead?
    return ((BytesWritable) obj).getBytes();
  } else {
    return new byte[0];
  }
}

public void write(Writable w) throws IOException {
  if (w instanceof TypedBytesWritable) {
    writeTypedBytes((TypedBytesWritable) w);
  } else if (w instanceof BytesWritable) {
    writeBytes((BytesWritable) w);
  } else if (w instanceof ByteWritable) {
    writeByte((ByteWritable) w);
  } else if (w instanceof BooleanWritable) {
    writeBoolean((BooleanWritable) w);
  } else if (w instanceof IntWritable) {
    writeInt((IntWritable) w);
  } else if (w instanceof VIntWritable) {
    writeVInt((VIntWritable) w);
  } else if (w instanceof LongWritable) {
    writeLong((LongWritable) w);
  } else if (w instanceof VLongWritable) {
    writeVLong((VLongWritable) w);
  } else if (w instanceof FloatWritable) {
    writeFloat((FloatWritable) w);
  } else if (w instanceof DoubleWritable) {
    writeDouble((DoubleWritable) w);
  } else if (w instanceof Text) {
    writeText((Text) w);
  } else if (w instanceof ArrayWritable) {
    writeArray((ArrayWritable) w);
  } else if (w instanceof MapWritable) {
    writeMap((MapWritable) w);
  } else if (w instanceof SortedMapWritable) {
    writeSortedMap((SortedMapWritable<?>) w);
  } else if (w instanceof Record) {
    writeRecord((Record) w);
  } else {
    writeWritable(w); // last resort
  }
}

public Class<? extends Writable> readType() throws IOException {
  Type type = in.readType();
  if (type == null) {
    return null;
  }
  switch (type) {
  case BYTES:
    return BytesWritable.class;
  case BYTE:
    return ByteWritable.class;
  case BOOL:
    return BooleanWritable.class;
  case INT:
    return VIntWritable.class;
  case LONG:
    return VLongWritable.class;
  case FLOAT:
    return FloatWritable.class;
  case DOUBLE:
    return DoubleWritable.class;
  case STRING:
    return Text.class;
  case VECTOR:
    return ArrayWritable.class;
  case MAP:
    return MapWritable.class;
  case WRITABLE:
    return Writable.class;
  default:
    throw new RuntimeException("unknown type");
  }
}

/**
 * Create new XdmValue from value type and Writables.
 *  
 */
public static XdmValue newValue(ValueType valueType, Object value) {
    if (value instanceof Text) {
        return ValueFactory.newValue(valueType, ((Text)value).toString());
    } else if (value instanceof BytesWritable) {
        return ValueFactory.newValue(valueType, ((BytesWritable)value).getBytes());
    } else if (value instanceof IntWritable) {
        return ValueFactory.newValue(valueType, ((IntWritable)value).get());
    } else if (value instanceof LongWritable) {
        return ValueFactory.newValue(valueType, ((LongWritable)value).get());
    } else if (value instanceof VIntWritable) {
        return ValueFactory.newValue(valueType, ((VIntWritable)value).get());
    } else if (value instanceof VLongWritable) {
        return ValueFactory.newValue(valueType, ((VLongWritable)value).get());
    } else if (value instanceof BooleanWritable) {
        return ValueFactory.newValue(valueType, ((BooleanWritable)value).get());
    } else if (value instanceof FloatWritable) {
        return ValueFactory.newValue(valueType, ((FloatWritable)value).get());
    } else if (value instanceof DoubleWritable) {
        return ValueFactory.newValue(valueType, ((DoubleWritable)value).get());
    } else if (value instanceof MarkLogicNode) {
        return ValueFactory.newValue(valueType, ((MarkLogicNode)value).get());
    } else {
        throw new UnsupportedOperationException("Value " +  
                value.getClass().getName() + " is unsupported.");
    }
}

public void write(Writable w) throws IOException {
  if (w instanceof TypedBytesWritable) {
    writeTypedBytes((TypedBytesWritable) w);
  } else if (w instanceof BytesWritable) {
    writeBytes((BytesWritable) w);
  } else if (w instanceof ByteWritable) {
    writeByte((ByteWritable) w);
  } else if (w instanceof BooleanWritable) {
    writeBoolean((BooleanWritable) w);
  } else if (w instanceof IntWritable) {
    writeInt((IntWritable) w);
  } else if (w instanceof VIntWritable) {
    writeVInt((VIntWritable) w);
  } else if (w instanceof LongWritable) {
    writeLong((LongWritable) w);
  } else if (w instanceof VLongWritable) {
    writeVLong((VLongWritable) w);
  } else if (w instanceof FloatWritable) {
    writeFloat((FloatWritable) w);
  } else if (w instanceof DoubleWritable) {
    writeDouble((DoubleWritable) w);
  } else if (w instanceof Text) {
    writeText((Text) w);
  } else if (w instanceof ArrayWritable) {
    writeArray((ArrayWritable) w);
  } else if (w instanceof MapWritable) {
    writeMap((MapWritable) w);
  } else if (w instanceof SortedMapWritable) {
    writeSortedMap((SortedMapWritable) w);
  } else if (w instanceof Record) {
    writeRecord((Record) w);
  } else {
    writeWritable(w); // last resort
  }
}

public Class<? extends Writable> readType() throws IOException {
  Type type = in.readType();
  if (type == null) {
    return null;
  }
  switch (type) {
  case BYTES:
    return BytesWritable.class;
  case BYTE:
    return ByteWritable.class;
  case BOOL:
    return BooleanWritable.class;
  case INT:
    return VIntWritable.class;
  case LONG:
    return VLongWritable.class;
  case FLOAT:
    return FloatWritable.class;
  case DOUBLE:
    return DoubleWritable.class;
  case STRING:
    return Text.class;
  case VECTOR:
    return ArrayWritable.class;
  case MAP:
    return MapWritable.class;
  case WRITABLE:
    return Writable.class;
  default:
    throw new RuntimeException("unknown type");
  }
}

public void reduce(LongWritable key, Iterable<VLongWritable> values, Context context) throws IOException, InterruptedException {

  int defCount = 0;

  refs.clear();
  for (VLongWritable type : values) {
    if (type.get() == -1) {
      defCount++;
    } else {
      refs.add(type.get());
    }
  }

  // TODO check for more than one def, should not happen

  if (defCount == 0 && refs.size() > 0) {
    // this is bad, found a node that is referenced but not defined. It must have been lost, emit some info about this node for debugging purposes.

    StringBuilder sb = new StringBuilder();
    String comma = "";
    for (Long ref : refs) {
      sb.append(comma);
      comma = ",";
      sb.append(String.format(Locale.getDefault(), "%016x", ref));
    }

    context.write(new Text(String.format(Locale.getDefault(), "%016x", key.get())), new Text(sb.toString()));
    context.getCounter(Counts.UNDEFINED).increment(1);

  } else if (defCount > 0 && refs.size() == 0) {
    // node is defined but not referenced
    context.getCounter(Counts.UNREFERENCED).increment(1);
  } else {
    // node is defined and referenced
    context.getCounter(Counts.REFERENCED).increment(1);
  }

}

public void start(Path outputDir, int numReducers, boolean concurrent) throws Exception {
  LOG.info("Running Verify with outputDir=" + outputDir +", numReducers=" + numReducers);

  DataStore<Long,CINode> store = DataStoreFactory.getDataStore(Long.class, CINode.class, new Configuration());

  job = new Job(getConf());

  if (!job.getConfiguration().get("io.serializations").contains("org.apache.hadoop.io.serializer.JavaSerialization")) {
    job.getConfiguration().set("io.serializations", job.getConfiguration().get("io.serializations") + ",org.apache.hadoop.io.serializer.JavaSerialization");
  }

  job.setJobName("Link Verifier");
  job.setNumReduceTasks(numReducers);
  job.setJarByClass(getClass());

  Query<Long,CINode> query = store.newQuery();
  if (!concurrent) {
    // no concurrency filtering, only need prev field
    query.setFields("prev");
  } else {
    readFlushed(job.getConfiguration());
  }

  GoraMapper.initMapperJob(job, query, store, LongWritable.class, VLongWritable.class, VerifyMapper.class, true);

  job.getConfiguration().setBoolean("mapred.map.tasks.speculative.execution", false);

  job.setReducerClass(VerifyReducer.class);
  job.setOutputFormatClass(TextOutputFormat.class);
  TextOutputFormat.setOutputPath(job, outputDir);

  store.close();

  job.submit();
}

public static Writable toWritable(Object object) {
    if (object == null) {
        return null; //return NullWritable.get();
    }
    if (object instanceof Writable) {
        return (Writable) object;
    }
    if (object instanceof String) {
        return new Text((String) object);
    }
    if (object instanceof Long) {
        return new VLongWritable((Long) object);
    }
    if (object instanceof Integer) {
        return new VIntWritable((Integer) object);
    }
    if (object instanceof Byte) {
        return new ByteWritable((Byte) object);
    }
    if (object instanceof Double) {
        return new DoubleWritable((Double) object);
    }
    if (object instanceof Float) {
        return new FloatWritable((Float) object);
    }
    if (object instanceof Boolean) {
        return new BooleanWritable((Boolean) object);
    }
    if (object instanceof byte[]) {
        return new BytesWritable((byte[]) object);
    }
    return new BytesWritable(object.toString().getBytes());
}

public static void updateObject(Writable obj, byte[] seed) {
  if (obj instanceof IntWritable) {
    ((IntWritable)obj).set(Ints.fromByteArray(seed));
  } else if (obj instanceof FloatWritable) {
    ((FloatWritable)obj).set(r.nextFloat());
  } else if (obj instanceof DoubleWritable) {
    ((DoubleWritable)obj).set(r.nextDouble());
  } else if (obj instanceof LongWritable) {
    ((LongWritable)obj).set(Longs.fromByteArray(seed));
  } else if (obj instanceof VIntWritable) {
    ((VIntWritable)obj).set(Ints.fromByteArray(seed));
  } else if (obj instanceof VLongWritable) {
    ((VLongWritable)obj).set(Longs.fromByteArray(seed));
  } else if (obj instanceof BooleanWritable) {
    ((BooleanWritable)obj).set(seed[0] % 2 == 1 ? true : false);
  } else if (obj instanceof Text) {
    ((Text)obj).set(BytesUtil.toStringBinary(seed));
  } else if (obj instanceof ByteWritable) {
    ((ByteWritable)obj).set(seed.length > 0 ? seed[0] : 0);
  } else if (obj instanceof BytesWritable) {
    ((BytesWritable)obj).set(seed, 0, seed.length);
  } else if (obj instanceof UTF8) {
    ((UTF8)obj).set(BytesUtil.toStringBinary(seed));
  } else if (obj instanceof MockValueClass) {
    ((MockValueClass)obj).set(seed);
  } else {
    throw new IllegalArgumentException("unknown writable: " +
                                       obj.getClass().getName());
  }
}

public static <VTYPE> byte[] toBytes(VTYPE obj) {
  final String className = obj.getClass().getName();
  if (className.equals(IntWritable.class.getName())) {
    return Ints.toByteArray(((IntWritable) obj).get());
  } else if (className.equals(FloatWritable.class.getName())) {
    return BytesUtil.toBytes(((FloatWritable) obj).get());
  } else if (className.equals(DoubleWritable.class.getName())) {
    return BytesUtil.toBytes(((DoubleWritable) obj).get());
  } else if (className.equals(LongWritable.class.getName())) {
    return Longs.toByteArray(((LongWritable) obj).get());
  } else if (className.equals(VIntWritable.class.getName())) {
    return Ints.toByteArray(((VIntWritable) obj).get());
  } else if (className.equals(VLongWritable.class.getName())) {
    return Longs.toByteArray(((VLongWritable) obj).get());
  } else if (className.equals(BooleanWritable.class.getName())) {
    return BytesUtil.toBytes(((BooleanWritable) obj).get());
  } else if (className.equals(Text.class.getName())) {
    return ((Text)obj).copyBytes();
  } else if (className.equals(ByteWritable.class.getName())) {
    return Ints.toByteArray((int) ((ByteWritable) obj).get());
  } else if (className.equals(BytesWritable.class.getName())) {
    // TODO: copyBytes instead?
    return ((BytesWritable) obj).getBytes();
  } else {
    return new byte[0];
  }
}

public void write(Writable w) throws IOException {
  if (w instanceof TypedBytesWritable) {
    writeTypedBytes((TypedBytesWritable) w);
  } else if (w instanceof BytesWritable) {
    writeBytes((BytesWritable) w);
  } else if (w instanceof ByteWritable) {
    writeByte((ByteWritable) w);
  } else if (w instanceof BooleanWritable) {
    writeBoolean((BooleanWritable) w);
  } else if (w instanceof IntWritable) {
    writeInt((IntWritable) w);
  } else if (w instanceof VIntWritable) {
    writeVInt((VIntWritable) w);
  } else if (w instanceof LongWritable) {
    writeLong((LongWritable) w);
  } else if (w instanceof VLongWritable) {
    writeVLong((VLongWritable) w);
  } else if (w instanceof FloatWritable) {
    writeFloat((FloatWritable) w);
  } else if (w instanceof DoubleWritable) {
    writeDouble((DoubleWritable) w);
  } else if (w instanceof Text) {
    writeText((Text) w);
  } else if (w instanceof ArrayWritable) {
    writeArray((ArrayWritable) w);
  } else if (w instanceof MapWritable) {
    writeMap((MapWritable) w);
  } else if (w instanceof SortedMapWritable) {
    writeSortedMap((SortedMapWritable) w);
  } else if (w instanceof Record) {
    writeRecord((Record) w);
  } else {
    writeWritable(w); // last resort
  }
}

public Class<? extends Writable> readType() throws IOException {
  Type type = in.readType();
  if (type == null) {
    return null;
  }
  switch (type) {
  case BYTES:
    return BytesWritable.class;
  case BYTE:
    return ByteWritable.class;
  case BOOL:
    return BooleanWritable.class;
  case INT:
    return VIntWritable.class;
  case LONG:
    return VLongWritable.class;
  case FLOAT:
    return FloatWritable.class;
  case DOUBLE:
    return DoubleWritable.class;
  case STRING:
    return Text.class;
  case VECTOR:
    return ArrayWritable.class;
  case MAP:
    return MapWritable.class;
  case WRITABLE:
    return Writable.class;
  default:
    throw new RuntimeException("unknown type");
  }
}

public void write(Writable w) throws IOException {
  if (w instanceof TypedBytesWritable) {
    writeTypedBytes((TypedBytesWritable) w);
  } else if (w instanceof BytesWritable) {
    writeBytes((BytesWritable) w);
  } else if (w instanceof ByteWritable) {
    writeByte((ByteWritable) w);
  } else if (w instanceof BooleanWritable) {
    writeBoolean((BooleanWritable) w);
  } else if (w instanceof IntWritable) {
    writeInt((IntWritable) w);
  } else if (w instanceof VIntWritable) {
    writeVInt((VIntWritable) w);
  } else if (w instanceof LongWritable) {
    writeLong((LongWritable) w);
  } else if (w instanceof VLongWritable) {
    writeVLong((VLongWritable) w);
  } else if (w instanceof FloatWritable) {
    writeFloat((FloatWritable) w);
  } else if (w instanceof DoubleWritable) {
    writeDouble((DoubleWritable) w);
  } else if (w instanceof Text) {
    writeText((Text) w);
  } else if (w instanceof ArrayWritable) {
    writeArray((ArrayWritable) w);
  } else if (w instanceof MapWritable) {
    writeMap((MapWritable) w);
  } else if (w instanceof SortedMapWritable) {
    writeSortedMap((SortedMapWritable) w);
  } else if (w instanceof Record) {
    writeRecord((Record) w);
  } else {
    writeWritable(w); // last resort
  }
}