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

public KeyValueWriter(Configuration conf, OutputStream output,
                      Class<K> kyClass, Class<V> valClass
                     ) throws IOException {
  keyClass = kyClass;
  valueClass = valClass;
  dataBuffer = new DataOutputBuffer();
  SerializationFactory serializationFactory
                                         = new SerializationFactory(conf);
  keySerializer
              = (Serializer<K>)serializationFactory.getSerializer(keyClass);
  keySerializer.open(dataBuffer);
  valueSerializer
            = (Serializer<V>)serializationFactory.getSerializer(valueClass);
  valueSerializer.open(dataBuffer);
  outputStream = new DataOutputStream(output);
}

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}

/**
 * This method is called to write the record that was most recently
 * served (before a call to the mark). Since the framework reads one
 * record in advance, to get this record, we serialize the current key
 * and value
 * @param out
 * @throws IOException
 */
private void writeFirstKeyValueBytes(DataOutputStream out) 
throws IOException {
  assert (getCurrentKey() != null && getCurrentValue() != null);
  WritableUtils.writeVInt(out, currentKeyLength);
  WritableUtils.writeVInt(out, currentValueLength);
  Serializer<KEYIN> keySerializer = 
    serializationFactory.getSerializer(keyClass);
  keySerializer.open(out);
  keySerializer.serialize(getCurrentKey());

  Serializer<VALUEIN> valueSerializer = 
    serializationFactory.getSerializer(valueClass);
  valueSerializer.open(out);
  valueSerializer.serialize(getCurrentValue());
}

/**
 * Make a copy of the writable object using serialization to a buffer
 * @param dst the object to copy from
 * @param src the object to copy into, which is destroyed
 * @throws IOException
 */
@SuppressWarnings("unchecked")
public static <T> T copy(Configuration conf, 
                              T src, T dst) throws IOException {
  CopyInCopyOutBuffer buffer = cloneBuffers.get();
  buffer.outBuffer.reset();
  SerializationFactory factory = getFactory(conf);
  Class<T> cls = (Class<T>) src.getClass();
  Serializer<T> serializer = factory.getSerializer(cls);
  serializer.open(buffer.outBuffer);
  serializer.serialize(src);
  buffer.moveData();
  Deserializer<T> deserializer = factory.getDeserializer(cls);
  deserializer.open(buffer.inBuffer);
  dst = deserializer.deserialize(dst);
  return dst;
}

public KeyValueWriter(Configuration conf, OutputStream output,
                      Class<K> kyClass, Class<V> valClass
                     ) throws IOException {
  keyClass = kyClass;
  valueClass = valClass;
  dataBuffer = new DataOutputBuffer();
  SerializationFactory serializationFactory
                                         = new SerializationFactory(conf);
  keySerializer
              = (Serializer<K>)serializationFactory.getSerializer(keyClass);
  keySerializer.open(dataBuffer);
  valueSerializer
            = (Serializer<V>)serializationFactory.getSerializer(valueClass);
  valueSerializer.open(dataBuffer);
  outputStream = new DataOutputStream(output);
}

private <K> K serDeser(K conf) throws Exception {
  SerializationFactory factory = new SerializationFactory(CONF);
  Serializer<K> serializer =
    factory.getSerializer(GenericsUtil.getClass(conf));
  Deserializer<K> deserializer =
    factory.getDeserializer(GenericsUtil.getClass(conf));

  DataOutputBuffer out = new DataOutputBuffer();
  serializer.open(out);
  serializer.serialize(conf);
  serializer.close();

  DataInputBuffer in = new DataInputBuffer();
  in.reset(out.getData(), out.getLength());
  deserializer.open(in);
  K after = deserializer.deserialize(null);
  deserializer.close();
  return after;
}

private <E> E makeCopyForPassByValue(Serialization<E> serialization,
                                      E obj) throws IOException {
  Serializer<E> ser =
    serialization.getSerializer(GenericsUtil.getClass(obj));
  Deserializer<E> deser =
    serialization.getDeserializer(GenericsUtil.getClass(obj));

  DataOutputBuffer dof = threadLocalDataOutputBuffer.get();

  dof.reset();
  ser.open(dof);
  ser.serialize(obj);
  ser.close();
  obj = ReflectionUtils.newInstance(GenericsUtil.getClass(obj),
                                    getChainJobConf());
  ByteArrayInputStream bais =
    new ByteArrayInputStream(dof.getData(), 0, dof.getLength());
  deser.open(bais);
  deser.deserialize(obj);
  deser.close();
  return obj;
}

/**
 * This method is called to write the record that was most recently
 * served (before a call to the mark). Since the framework reads one
 * record in advance, to get this record, we serialize the current key
 * and value
 * @param out
 * @throws IOException
 */
private void writeFirstKeyValueBytes(DataOutputStream out) 
throws IOException {
  assert (getCurrentKey() != null && getCurrentValue() != null);
  WritableUtils.writeVInt(out, currentKeyLength);
  WritableUtils.writeVInt(out, currentValueLength);
  Serializer<KEYIN> keySerializer = 
    serializationFactory.getSerializer(keyClass);
  keySerializer.open(out);
  keySerializer.serialize(getCurrentKey());

  Serializer<VALUEIN> valueSerializer = 
    serializationFactory.getSerializer(valueClass);
  valueSerializer.open(out);
  valueSerializer.serialize(getCurrentValue());
}

@Test
public void testSerializer() throws Exception {
   WebPage originalWebPage = new WebPage(new URL("http://www.jboss.org"), "opensource", 10L);

   ByteArrayOutputStream baos = new ByteArrayOutputStream();

   JBossMarshallerSerialization<WebPage> marshallerSerialization = new JBossMarshallerSerialization<>();
   Serializer<WebPage> serializer = marshallerSerialization.getSerializer(WebPage.class);
   serializer.open(baos);
   serializer.serialize(originalWebPage);
   serializer.close();

   Deserializer<WebPage> deserializer = marshallerSerialization.getDeserializer(WebPage.class);
   deserializer.open(new ByteArrayInputStream(baos.toByteArray()));
   WebPage deserializedWebPage = deserializer.deserialize(null);
   deserializer.close();

   assertEquals(deserializedWebPage, originalWebPage);
}