Author: tomwhite
Date: Thu Sep 3 01:36:33 2009
New Revision: 810756
URL: http://svn.apache.org/viewvc?rev=810756&view=rev
Log:
HADOOP-6165. Add metadata to Serializations.
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerBase.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacyDeserializer.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerializer.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationBase.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializerBase.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroGenericSerialization.java
Modified:
hadoop/common/trunk/CHANGES.txt
hadoop/common/trunk/src/java/core-default.xml
hadoop/common/trunk/src/java/org/apache/hadoop/io/DefaultStringifier.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/SequenceFile.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Deserializer.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerComparator.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationFactory.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serializer.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSerialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSpecificSerialization.java
hadoop/common/trunk/src/java/org/apache/hadoop/util/ReflectionUtils.java
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/SerializationTestUtil.java
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/avro/TestAvroSerialization.java
Modified: hadoop/common/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/CHANGES.txt?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
--- hadoop/common/trunk/CHANGES.txt (original)
+++ hadoop/common/trunk/CHANGES.txt Thu Sep 3 01:36:33 2009
@@ -166,6 +166,8 @@
the io package and makes it available to other users (MAPREDUCE-318).
(Jothi Padmanabhan via ddas)
+ HADOOP-6165. Add metadata to Serializations. (tomwhite)
+
IMPROVEMENTS
HADOOP-4565. Added CombineFileInputFormat to use data locality information
Modified: hadoop/common/trunk/src/java/core-default.xml
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/core-default.xml?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/core-default.xml (original)
+++ hadoop/common/trunk/src/java/core-default.xml Thu Sep 3 01:36:33 2009
@@ -101,7 +101,7 @@
<property>
<name>io.serializations</name>
-
<value>org.apache.hadoop.io.serializer.WritableSerialization,org.apache.hadoop.io.serializer.avro.AvroSpecificSerialization,org.apache.hadoop.io.serializer.avro.AvroReflectSerialization</value>
+
<value>org.apache.hadoop.io.serializer.WritableSerialization,org.apache.hadoop.io.serializer.avro.AvroSpecificSerialization,org.apache.hadoop.io.serializer.avro.AvroReflectSerialization,org.apache.hadoop.io.serializer.avro.AvroGenericSerialization</value>
<description>A list of serialization classes that can be used for
obtaining serializers and deserializers.</description>
</property>
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/DefaultStringifier.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/DefaultStringifier.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/io/DefaultStringifier.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/io/DefaultStringifier.java
Thu Sep 3 01:36:33 2009
@@ -21,20 +21,21 @@
import java.io.IOException;
import java.nio.charset.UnsupportedCharsetException;
import java.util.ArrayList;
+import java.util.Map;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.serializer.Deserializer;
-import org.apache.hadoop.io.serializer.Serialization;
+import org.apache.hadoop.io.serializer.DeserializerBase;
+import org.apache.hadoop.io.serializer.SerializationBase;
import org.apache.hadoop.io.serializer.SerializationFactory;
-import org.apache.hadoop.io.serializer.Serializer;
+import org.apache.hadoop.io.serializer.SerializerBase;
import org.apache.hadoop.util.GenericsUtil;
/**
* DefaultStringifier is the default implementation of the {...@link
Stringifier}
* interface which stringifies the objects using base64 encoding of the
- * serialized version of the objects. The {...@link Serializer} and
- * {...@link Deserializer} are obtained from the {...@link
SerializationFactory}.
+ * serialized version of the objects. The {...@link SerializerBase} and
+ * {...@link DeserializerBase} are obtained from the {...@link
SerializationFactory}.
* <br>
* DefaultStringifier offers convenience methods to store/load objects to/from
* the configuration.
@@ -45,9 +46,9 @@
private static final String SEPARATOR = ",";
- private Serializer<T> serializer;
+ private SerializerBase<T> serializer;
- private Deserializer<T> deserializer;
+ private DeserializerBase<T> deserializer;
private DataInputBuffer inBuf;
@@ -56,8 +57,9 @@
public DefaultStringifier(Configuration conf, Class<T> c) {
SerializationFactory factory = new SerializationFactory(conf);
- this.serializer = factory.getSerializer(c);
- this.deserializer = factory.getDeserializer(c);
+ Map<String, String> metadata = SerializationBase.getMetadataFromClass(c);
+ this.serializer = factory.getSerializer(metadata);
+ this.deserializer = factory.getDeserializer(metadata);
this.inBuf = new DataInputBuffer();
this.outBuf = new DataOutputBuffer();
try {
@@ -102,7 +104,7 @@
* @param item the object to be stored
* @param keyName the name of the key to use
* @throws IOException : forwards Exceptions from the underlying
- * {...@link Serialization} classes.
+ * {...@link SerializationBase} classes.
*/
public static <K> void store(Configuration conf, K item, String keyName)
throws IOException {
@@ -122,7 +124,7 @@
* @param itemClass the class of the item
* @return restored object
* @throws IOException : forwards Exceptions from the underlying
- * {...@link Serialization} classes.
+ * {...@link SerializationBase} classes.
*/
public static <K> K load(Configuration conf, String keyName,
Class<K> itemClass) throws IOException {
@@ -145,7 +147,7 @@
* @param keyName the name of the key to use
* @throws IndexOutOfBoundsException if the items array is empty
* @throws IOException : forwards Exceptions from the underlying
- * {...@link Serialization} classes.
+ * {...@link SerializationBase} classes.
*/
public static <K> void storeArray(Configuration conf, K[] items,
String keyName) throws IOException {
@@ -173,7 +175,7 @@
* @param itemClass the class of the item
* @return restored object
* @throws IOException : forwards Exceptions from the underlying
- * {...@link Serialization} classes.
+ * {...@link SerializationBase} classes.
*/
public static <K> K[] loadArray(Configuration conf, String keyName,
Class<K> itemClass) throws IOException {
Modified: hadoop/common/trunk/src/java/org/apache/hadoop/io/SequenceFile.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/SequenceFile.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/io/SequenceFile.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/io/SequenceFile.java Thu Sep
3 01:36:33 2009
@@ -33,9 +33,10 @@
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.zlib.ZlibFactory;
-import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.DeserializerBase;
+import org.apache.hadoop.io.serializer.SerializationBase;
+import org.apache.hadoop.io.serializer.SerializerBase;
import org.apache.hadoop.io.serializer.SerializationFactory;
-import org.apache.hadoop.io.serializer.Serializer;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.Progress;
@@ -705,6 +706,14 @@
return new TreeMap<Text, Text>(this.theMetadata);
}
+ public Map<String, String> getMetadataAsStringMap() {
+ Map<String, String> map = new HashMap<String, String>();
+ for (Map.Entry<Text, Text> entry : theMetadata.entrySet()) {
+ map.put(entry.getKey().toString(), entry.getValue().toString());
+ }
+ return map;
+ }
+
public void write(DataOutput out) throws IOException {
out.writeInt(this.theMetadata.size());
Iterator<Map.Entry<Text, Text>> iter =
@@ -801,9 +810,9 @@
Metadata metadata = null;
Compressor compressor = null;
- protected Serializer keySerializer;
- protected Serializer uncompressedValSerializer;
- protected Serializer compressedValSerializer;
+ protected SerializerBase keySerializer;
+ protected SerializerBase uncompressedValSerializer;
+ protected SerializerBase compressedValSerializer;
// Insert a globally unique 16-byte value every few entries, so that one
// can seek into the middle of a file and then synchronize with record
@@ -914,9 +923,10 @@
this.codec = codec;
this.metadata = metadata;
SerializationFactory serializationFactory = new
SerializationFactory(conf);
- this.keySerializer = serializationFactory.getSerializer(keyClass);
+ this.keySerializer = getSerializer(serializationFactory, keyClass,
metadata);
this.keySerializer.open(buffer);
- this.uncompressedValSerializer =
serializationFactory.getSerializer(valClass);
+ this.uncompressedValSerializer = getSerializer(serializationFactory,
+ valClass, metadata);
this.uncompressedValSerializer.open(buffer);
if (this.codec != null) {
ReflectionUtils.setConf(this.codec, this.conf);
@@ -924,11 +934,20 @@
this.deflateFilter = this.codec.createOutputStream(buffer, compressor);
this.deflateOut =
new DataOutputStream(new BufferedOutputStream(deflateFilter));
- this.compressedValSerializer =
serializationFactory.getSerializer(valClass);
+ this.compressedValSerializer = getSerializer(serializationFactory,
+ valClass, metadata);
this.compressedValSerializer.open(deflateOut);
}
}
+ @SuppressWarnings("unchecked")
+ private SerializerBase getSerializer(SerializationFactory sf, Class c,
+ Metadata metadata) {
+ Map<String, String> stringMetadata = metadata.getMetadataAsStringMap();
+ stringMetadata.put(SerializationBase.CLASS_KEY, c.getName());
+ return sf.getSerializer(stringMetadata);
+ }
+
/** Returns the class of keys in this file. */
public Class getKeyClass() { return keyClass; }
@@ -1412,8 +1431,8 @@
private DataInputStream valIn = null;
private Decompressor valDecompressor = null;
- private Deserializer keyDeserializer;
- private Deserializer valDeserializer;
+ private DeserializerBase keyDeserializer;
+ private DeserializerBase valDeserializer;
/** Open the named file. */
public Reader(FileSystem fs, Path file, Configuration conf)
@@ -1563,21 +1582,24 @@
SerializationFactory serializationFactory =
new SerializationFactory(conf);
this.keyDeserializer =
- getDeserializer(serializationFactory, getKeyClass());
+ getDeserializer(serializationFactory, getKeyClass(), metadata);
if (!blockCompressed) {
this.keyDeserializer.open(valBuffer);
} else {
this.keyDeserializer.open(keyIn);
}
this.valDeserializer =
- getDeserializer(serializationFactory, getValueClass());
+ getDeserializer(serializationFactory, getValueClass(), metadata);
this.valDeserializer.open(valIn);
}
}
@SuppressWarnings("unchecked")
- private Deserializer getDeserializer(SerializationFactory sf, Class c) {
- return sf.getDeserializer(c);
+ private DeserializerBase getDeserializer(SerializationFactory sf, Class c,
+ Metadata metadata) {
+ Map<String, String> stringMetadata = metadata.getMetadataAsStringMap();
+ stringMetadata.put(SerializationBase.CLASS_KEY, c.getName());
+ return sf.getDeserializer(stringMetadata);
}
/** Close the file. */
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Deserializer.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Deserializer.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Deserializer.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Deserializer.java
Thu Sep 3 01:36:33 2009
@@ -34,6 +34,7 @@
* </p>
* @param <T>
*/
+...@deprecated
public interface Deserializer<T> {
/**
* <p>Prepare the deserializer for reading.</p>
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerBase.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerBase.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerBase.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerBase.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.io.serializer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.conf.Configured;
+
+public abstract class DeserializerBase<T> extends Configured
+ implements Closeable, Deserializer<T> {
+
+ /**
+ * <p>Prepare the deserializer for reading.</p>
+ */
+ public abstract void open(InputStream in) throws IOException;
+
+ /**
+ * <p>
+ * Deserialize the next object from the underlying input stream.
+ * If the object <code>t</code> is non-null then this deserializer
+ * <i>may</i> set its internal state to the next object read from the input
+ * stream. Otherwise, if the object <code>t</code> is null a new
+ * deserialized object will be created.
+ * </p>
+ * @return the deserialized object
+ */
+ public abstract T deserialize(T t) throws IOException;
+
+}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerComparator.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerComparator.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerComparator.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/DeserializerComparator.java
Thu Sep 3 01:36:33 2009
@@ -52,6 +52,13 @@
this.deserializer.open(buffer);
}
+ protected DeserializerComparator(DeserializerBase<T> deserializer)
+ throws IOException {
+
+ this.deserializer = deserializer;
+ this.deserializer.open(buffer);
+ }
+
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
try {
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacyDeserializer.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacyDeserializer.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacyDeserializer.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacyDeserializer.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.io.serializer;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+...@suppresswarnings("deprecation")
+class LegacyDeserializer<T> extends DeserializerBase<T> {
+
+ private Deserializer<T> deserializer;
+
+ public LegacyDeserializer(Deserializer<T> deserializer) {
+ this.deserializer = deserializer;
+ }
+
+ @Override
+ public void open(InputStream in) throws IOException {
+ deserializer.open(in);
+ }
+
+ @Override
+ public T deserialize(T t) throws IOException {
+ return deserializer.deserialize(t);
+ }
+
+ @Override
+ public void close() throws IOException {
+ deserializer.close();
+ }
+
+}
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerialization.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerialization.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerialization.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.io.serializer;
+
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+
+/**
+ * <p>
+ * Wraps a legacy {...@link Serialization} as a {...@link SerializationBase}.
+ * </p>
+ *
+ * @param <T>
+ */
+...@suppresswarnings("deprecation")
+class LegacySerialization<T> extends SerializationBase<T> {
+
+ private Serialization<T> serialization;
+
+ public LegacySerialization(Serialization<T> serialization,
+ Configuration conf) {
+ this.serialization = serialization;
+ setConf(conf);
+ }
+
+ Serialization<T> getUnderlyingSerialization() {
+ return serialization;
+ }
+
+ @Deprecated
+ @Override
+ public boolean accept(Class<?> c) {
+ return serialization.accept(c);
+ }
+
+ @Deprecated
+ @Override
+ public Deserializer<T> getDeserializer(Class<T> c) {
+ return serialization.getDeserializer(c);
+ }
+
+ @Deprecated
+ @Override
+ public Serializer<T> getSerializer(Class<T> c) {
+ return serialization.getSerializer(c);
+ }
+
+ @Override
+ public boolean accept(Map<String, String> metadata) {
+ Class<?> c = getClassFromMetadata(metadata);
+ return accept(c);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public SerializerBase<T> getSerializer(Map<String, String> metadata) {
+ Class<T> c = (Class<T>) getClassFromMetadata(metadata);
+ return new LegacySerializer<T>(getSerializer(c));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public DeserializerBase<T> getDeserializer(Map<String, String> metadata) {
+ Class<T> c = (Class<T>) getClassFromMetadata(metadata);
+ return new LegacyDeserializer<T>(getDeserializer(c));
+ }
+
+}
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerializer.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerializer.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerializer.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/LegacySerializer.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.io.serializer;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.Map;
+
+...@suppresswarnings("deprecation")
+class LegacySerializer<T> extends SerializerBase<T> {
+
+ private Serializer<T> serializer;
+
+ public LegacySerializer(Serializer<T> serializer) {
+ this.serializer = serializer;
+ }
+
+ @Override
+ public void open(OutputStream out) throws IOException {
+ serializer.open(out);
+ }
+
+ @Override
+ public void serialize(T t) throws IOException {
+ serializer.serialize(t);
+ }
+
+ @Override
+ public void close() throws IOException {
+ serializer.close();
+ }
+
+ @Override
+ public Map<String, String> getMetadata() throws IOException {
+ return Collections.<String, String>emptyMap();
+ }
+
+}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serialization.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serialization.java
Thu Sep 3 01:36:33 2009
@@ -24,6 +24,7 @@
* </p>
* @param <T>
*/
+...@deprecated
public interface Serialization<T> {
/**
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationBase.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationBase.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationBase.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationBase.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.io.serializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configured;
+
+/**
+ * <p>
+ * Encapsulates a {...@link SerializerBase}/{...@link DeserializerBase} pair.
+ * </p>
+ *
+ * @param <T>
+ */
+public abstract class SerializationBase<T> extends Configured
+ implements Serialization<T> {
+
+ public static final String SERIALIZATION_KEY = "Serialization-Class";
+ public static final String CLASS_KEY = "Serialized-Class";
+
+ public static Map<String, String> getMetadataFromClass(Class<?> c) {
+ Map<String, String> metadata = new HashMap<String, String>();
+ metadata.put(CLASS_KEY, c.getName());
+ return metadata;
+ }
+
+ @Deprecated
+ @Override
+ public boolean accept(Class<?> c) {
+ return accept(getMetadataFromClass(c));
+ }
+
+ @Deprecated
+ @Override
+ public Deserializer<T> getDeserializer(Class<T> c) {
+ return getDeserializer(getMetadataFromClass(c));
+ }
+
+ @Deprecated
+ @Override
+ public Serializer<T> getSerializer(Class<T> c) {
+ return getSerializer(getMetadataFromClass(c));
+ }
+
+ /**
+ * Allows clients to test whether this {...@link SerializationBase} supports
the
+ * given metadata.
+ */
+ public abstract boolean accept(Map<String, String> metadata);
+
+ /**
+ * @return a {...@link SerializerBase} for the given metadata.
+ */
+ public abstract SerializerBase<T> getSerializer(Map<String, String>
metadata);
+
+ /**
+ * @return a {...@link DeserializerBase} for the given metadata.
+ */
+ public abstract DeserializerBase<T> getDeserializer(
+ Map<String, String> metadata);
+
+ protected Class<?> getClassFromMetadata(Map<String, String> metadata) {
+ String classname = metadata.get(CLASS_KEY);
+ if (classname == null) {
+ return null;
+ }
+ try {
+ return getConf().getClassByName(classname);
+ } catch (ClassNotFoundException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationFactory.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationFactory.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationFactory.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializationFactory.java
Thu Sep 3 01:36:33 2009
@@ -20,11 +20,13 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.io.serializer.avro.AvroGenericSerialization;
import org.apache.hadoop.io.serializer.avro.AvroReflectSerialization;
import org.apache.hadoop.io.serializer.avro.AvroSpecificSerialization;
import org.apache.hadoop.util.ReflectionUtils;
@@ -32,7 +34,7 @@
/**
* <p>
- * A factory for {...@link Serialization}s.
+ * A factory for {...@link SerializationBase}s.
* </p>
*/
public class SerializationFactory extends Configured {
@@ -40,7 +42,10 @@
private static final Log LOG =
LogFactory.getLog(SerializationFactory.class.getName());
- private List<Serialization<?>> serializations = new
ArrayList<Serialization<?>>();
+ private List<SerializationBase<?>> serializations =
+ new ArrayList<SerializationBase<?>>();
+ private List<SerializationBase<?>> legacySerializations =
+ new ArrayList<SerializationBase<?>>();
/**
* <p>
@@ -54,7 +59,8 @@
for (String serializerName : conf.getStrings("io.serializations",
new String[]{WritableSerialization.class.getName(),
AvroSpecificSerialization.class.getName(),
- AvroReflectSerialization.class.getName()})) {
+ AvroReflectSerialization.class.getName(),
+ AvroGenericSerialization.class.getName()})) {
add(conf, serializerName);
}
}
@@ -62,30 +68,62 @@
@SuppressWarnings("unchecked")
private void add(Configuration conf, String serializationName) {
try {
-
- Class<? extends Serialization> serializionClass =
- (Class<? extends Serialization>)
conf.getClassByName(serializationName);
- serializations.add((Serialization)
- ReflectionUtils.newInstance(serializionClass, getConf()));
+ Class<?> serializationClass = conf.getClassByName(serializationName);
+ if (SerializationBase.class.isAssignableFrom(serializationClass)) {
+ serializations.add((SerializationBase)
+ ReflectionUtils.newInstance(serializationClass, getConf()));
+ } else if (Serialization.class.isAssignableFrom(serializationClass)) {
+ Serialization serialization = (Serialization)
+ ReflectionUtils.newInstance(serializationClass, getConf());
+ legacySerializations.add(new LegacySerialization(serialization,
+ getConf()));
+ } else {
+ LOG.warn("Serialization class " + serializationName + " is not an " +
+ "instance of Serialization or BaseSerialization.");
+ }
} catch (ClassNotFoundException e) {
- LOG.warn("Serilization class not found: " +
+ LOG.warn("Serialization class not found: " +
StringUtils.stringifyException(e));
}
}
+ @Deprecated
public <T> Serializer<T> getSerializer(Class<T> c) {
return getSerialization(c).getSerializer(c);
}
+ @Deprecated
public <T> Deserializer<T> getDeserializer(Class<T> c) {
return getSerialization(c).getDeserializer(c);
}
- @SuppressWarnings("unchecked")
+ @Deprecated
public <T> Serialization<T> getSerialization(Class<T> c) {
- for (Serialization serialization : serializations) {
- if (serialization.accept(c)) {
- return (Serialization<T>) serialization;
+ return getSerialization(SerializationBase.getMetadataFromClass(c));
+ }
+
+ public <T> SerializerBase<T> getSerializer(Map<String, String> metadata) {
+ SerializationBase<T> serialization = getSerialization(metadata);
+ return serialization.getSerializer(metadata);
+ }
+
+ public <T> DeserializerBase<T> getDeserializer(Map<String, String> metadata)
{
+ SerializationBase<T> serialization = getSerialization(metadata);
+ return serialization.getDeserializer(metadata);
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T> SerializationBase<T> getSerialization(Map<String, String>
metadata) {
+ for (SerializationBase serialization : serializations) {
+ if (serialization.accept(metadata)) {
+ return (SerializationBase<T>) serialization;
+ }
+ }
+ // Look in the legacy serializations last, since they ignore
+ // non-class metadata
+ for (SerializationBase serialization : legacySerializations) {
+ if (serialization.accept(metadata)) {
+ return (SerializationBase<T>) serialization;
}
}
return null;
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serializer.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serializer.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serializer.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/Serializer.java
Thu Sep 3 01:36:33 2009
@@ -34,6 +34,7 @@
* </p>
* @param <T>
*/
+...@deprecated
public interface Serializer<T> {
/**
* <p>Prepare the serializer for writing.</p>
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializerBase.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializerBase.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializerBase.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/SerializerBase.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.io.serializer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configured;
+
+public abstract class SerializerBase<T> extends Configured
+ implements Closeable, Serializer<T> {
+
+ /**
+ * <p>Prepare the serializer for writing.</p>
+ */
+ public abstract void open(OutputStream out) throws IOException;
+
+ /**
+ * <p>Serialize <code>t</code> to the underlying output stream.</p>
+ */
+ public abstract void serialize(T t) throws IOException;
+
+ public abstract Map<String, String> getMetadata() throws IOException;
+
+}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/WritableSerialization.java
Thu Sep 3 01:36:33 2009
@@ -23,22 +23,20 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
/**
- * A {...@link Serialization} for {...@link Writable}s that delegates to
+ * A {...@link SerializationBase} for {...@link Writable}s that delegates to
* {...@link Writable#write(java.io.DataOutput)} and
* {...@link Writable#readFields(java.io.DataInput)}.
*/
-public class WritableSerialization extends Configured
- implements Serialization<Writable> {
+public class WritableSerialization extends SerializationBase<Writable> {
- static class WritableDeserializer extends Configured
- implements Deserializer<Writable> {
+ static class WritableDeserializer extends DeserializerBase<Writable> {
private Class<?> writableClass;
private DataInputStream dataIn;
@@ -48,6 +46,7 @@
this.writableClass = c;
}
+ @Override
public void open(InputStream in) {
if (in instanceof DataInputStream) {
dataIn = (DataInputStream) in;
@@ -56,6 +55,7 @@
}
}
+ @Override
public Writable deserialize(Writable w) throws IOException {
Writable writable;
if (w == null) {
@@ -68,16 +68,23 @@
return writable;
}
+ @Override
public void close() throws IOException {
dataIn.close();
}
}
- static class WritableSerializer implements Serializer<Writable> {
-
+ static class WritableSerializer extends SerializerBase<Writable> {
+
+ private Map<String, String> metadata;
private DataOutputStream dataOut;
+ public WritableSerializer(Map<String, String> metadata) {
+ this.metadata = metadata;
+ }
+
+ @Override
public void open(OutputStream out) {
if (out instanceof DataOutputStream) {
dataOut = (DataOutputStream) out;
@@ -86,26 +93,41 @@
}
}
+ @Override
public void serialize(Writable w) throws IOException {
w.write(dataOut);
}
+ @Override
public void close() throws IOException {
dataOut.close();
}
- }
+ @Override
+ public Map<String, String> getMetadata() throws IOException {
+ return metadata;
+ }
- public boolean accept(Class<?> c) {
- return Writable.class.isAssignableFrom(c);
}
- public Deserializer<Writable> getDeserializer(Class<Writable> c) {
- return new WritableDeserializer(getConf(), c);
+ @Override
+ public boolean accept(Map<String, String> metadata) {
+ if (getClass().getName().equals(metadata.get(SERIALIZATION_KEY))) {
+ return true;
+ }
+ Class<?> c = getClassFromMetadata(metadata);
+ return c == null ? false : Writable.class.isAssignableFrom(c);
}
- public Serializer<Writable> getSerializer(Class<Writable> c) {
- return new WritableSerializer();
+ @Override
+ public SerializerBase<Writable> getSerializer(Map<String, String> metadata) {
+ return new WritableSerializer(metadata);
+ }
+
+ @Override
+ public DeserializerBase<Writable> getDeserializer(Map<String, String>
metadata) {
+ Class<?> c = getClassFromMetadata(metadata);
+ return new WritableDeserializer(getConf(), c);
}
}
Added:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroGenericSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroGenericSerialization.java?rev=810756&view=auto
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroGenericSerialization.java
(added)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroGenericSerialization.java
Thu Sep 3 01:36:33 2009
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.io.serializer.avro;
+
+import java.util.Map;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.io.DatumWriter;
+import org.apache.hadoop.io.serializer.SerializationBase;
+
+/**
+ * Serialization for Avro Generic classes. For a class to be accepted by this
+ * serialization it must have metadata with key
+ * {...@link SerializationBase#SERIALIZATION_KEY} set to {...@link
AvroGenericSerialization}'s
+ * fully-qualified classname.
+ * The schema used is the one set by {...@link
AvroSerialization#AVRO_SCHEMA_KEY}.
+ */
+...@suppresswarnings("unchecked")
+public class AvroGenericSerialization extends AvroSerialization<Object> {
+
+ @Override
+ public boolean accept(Map<String, String> metadata) {
+ return metadata.get(AVRO_SCHEMA_KEY) != null;
+ }
+
+ @Override
+ protected DatumReader getReader(Map<String, String> metadata) {
+ Schema schema = Schema.parse(metadata.get(AVRO_SCHEMA_KEY));
+ return new GenericDatumReader<Object>(schema);
+ }
+
+ @Override
+ protected Schema getSchema(Object t, Map<String, String> metadata) {
+ String jsonSchema = metadata.get(AVRO_SCHEMA_KEY);
+ return jsonSchema != null ? Schema.parse(jsonSchema) :
GenericData.induce(t);
+ }
+
+ @Override
+ protected DatumWriter getWriter(Map<String, String> metadata) {
+ return new GenericDatumWriter<Object>();
+ }
+
+}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroReflectSerialization.java
Thu Sep 3 01:36:33 2009
@@ -19,6 +19,7 @@
package org.apache.hadoop.io.serializer.avro;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import org.apache.avro.Schema;
@@ -27,6 +28,7 @@
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;
+import org.apache.avro.specific.SpecificRecord;
/**
* Serialization for Avro Reflect classes. For a class to be accepted by this
@@ -47,10 +49,18 @@
private Set<String> packages;
- public synchronized boolean accept(Class<?> c) {
+ @Override
+ public synchronized boolean accept(Map<String, String> metadata) {
if (packages == null) {
getPackages();
}
+ if (getClass().getName().equals(metadata.get(SERIALIZATION_KEY))) {
+ return true;
+ }
+ Class<?> c = getClassFromMetadata(metadata);
+ if (c == null) {
+ return false;
+ }
return AvroReflectSerializable.class.isAssignableFrom(c) ||
packages.contains(c.getPackage().getName());
}
@@ -65,8 +75,11 @@
}
}
- protected DatumReader getReader(Class<Object> clazz) {
+ @Override
+ protected DatumReader getReader(Map<String, String> metadata) {
try {
+ Class<SpecificRecord> clazz = (Class<SpecificRecord>)
+ getClassFromMetadata(metadata);
String prefix =
((clazz.getEnclosingClass() == null
|| "null".equals(clazz.getEnclosingClass().getName())) ?
@@ -78,11 +91,13 @@
}
}
- protected Schema getSchema(Object t) {
+ @Override
+ protected Schema getSchema(Object t, Map<String, String> metadata) {
return ReflectData.getSchema(t.getClass());
}
- protected DatumWriter getWriter(Class<Object> clazz) {
+ @Override
+ protected DatumWriter getWriter(Map<String, String> metadata) {
return new ReflectDatumWriter();
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSerialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSerialization.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSerialization.java
Thu Sep 3 01:36:33 2009
@@ -21,92 +21,105 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.Map;
import org.apache.avro.Schema;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
-import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.io.serializer.Deserializer;
-import org.apache.hadoop.io.serializer.Serialization;
-import org.apache.hadoop.io.serializer.Serializer;
+import org.apache.hadoop.io.serializer.DeserializerBase;
+import org.apache.hadoop.io.serializer.SerializationBase;
+import org.apache.hadoop.io.serializer.SerializerBase;
/**
* Base class for providing serialization to Avro types.
*/
-public abstract class AvroSerialization<T> extends Configured
- implements Serialization<T>{
+public abstract class AvroSerialization<T> extends SerializationBase<T> {
+
+ public static final String AVRO_SCHEMA_KEY = "Avro-Schema";
- public Deserializer<T> getDeserializer(Class<T> c) {
- return new AvroDeserializer(c);
+ public DeserializerBase<T> getDeserializer(Map<String, String> metadata) {
+ return new AvroDeserializer(metadata);
}
- public Serializer<T> getSerializer(Class<T> c) {
- return new AvroSerializer(c);
+ public SerializerBase<T> getSerializer(Map<String, String> metadata) {
+ return new AvroSerializer(metadata);
}
/**
- * Return an Avro Schema instance for the given class.
+ * Return an Avro Schema instance for the given class and metadata.
*/
- protected abstract Schema getSchema(T t);
+ protected abstract Schema getSchema(T t, Map<String, String> metadata);
/**
- * Create and return Avro DatumWriter for the given class.
+ * Create and return Avro DatumWriter for the given metadata.
*/
- protected abstract DatumWriter<T> getWriter(Class<T> clazz);
+ protected abstract DatumWriter<T> getWriter(Map<String, String> metadata);
/**
- * Create and return Avro DatumReader for the given class.
+ * Create and return Avro DatumReader for the given metadata.
*/
- protected abstract DatumReader<T> getReader(Class<T> clazz);
+ protected abstract DatumReader<T> getReader(Map<String, String> metadata);
- class AvroSerializer implements Serializer<T> {
+ class AvroSerializer extends SerializerBase<T> {
+ private Map<String, String> metadata;
private DatumWriter<T> writer;
private BinaryEncoder encoder;
private OutputStream outStream;
- protected Class<T> clazz;
- AvroSerializer(Class<T> clazz) {
- writer = getWriter(clazz);
+ AvroSerializer(Map<String, String> metadata) {
+ this.metadata = metadata;
+ writer = getWriter(metadata);
}
+ @Override
public void close() throws IOException {
encoder.flush();
outStream.close();
}
+ @Override
public void open(OutputStream out) throws IOException {
outStream = out;
encoder = new BinaryEncoder(out);
}
+ @Override
public void serialize(T t) throws IOException {
- writer.setSchema(getSchema(t));
+ writer.setSchema(getSchema(t, metadata));
writer.write(t, encoder);
}
+ @Override
+ public Map<String, String> getMetadata() throws IOException {
+ return metadata;
+ }
+
}
- class AvroDeserializer implements Deserializer<T> {
+ class AvroDeserializer extends DeserializerBase<T> {
private DatumReader<T> reader;
private BinaryDecoder decoder;
private InputStream inStream;
- AvroDeserializer(Class<T> clazz) {
- this.reader = getReader(clazz);
+ AvroDeserializer(Map<String, String> metadata) {
+ this.reader = getReader(metadata);
}
+ @Override
public void close() throws IOException {
inStream.close();
}
+ @Override
public T deserialize(T t) throws IOException {
return reader.read(t, decoder);
}
+ @Override
public void open(InputStream in) throws IOException {
inStream = in;
decoder = new BinaryDecoder(in);
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSpecificSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSpecificSerialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSpecificSerialization.java
(original)
+++
hadoop/common/trunk/src/java/org/apache/hadoop/io/serializer/avro/AvroSpecificSerialization.java
Thu Sep 3 01:36:33 2009
@@ -18,7 +18,10 @@
package org.apache.hadoop.io.serializer.avro;
+import java.util.Map;
+
import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
@@ -33,23 +36,33 @@
public class AvroSpecificSerialization
extends AvroSerialization<SpecificRecord>{
- public boolean accept(Class<?> c) {
- return SpecificRecord.class.isAssignableFrom(c);
+ @Override
+ public boolean accept(Map<String, String> metadata) {
+ if (getClass().getName().equals(metadata.get(SERIALIZATION_KEY))) {
+ return true;
+ }
+ Class<?> c = getClassFromMetadata(metadata);
+ return c == null ? false : SpecificRecord.class.isAssignableFrom(c);
}
- protected DatumReader getReader(Class<SpecificRecord> clazz) {
+ @Override
+ protected DatumReader getReader(Map<String, String> metadata) {
try {
+ Class<SpecificRecord> clazz = (Class<SpecificRecord>)
+ getClassFromMetadata(metadata);
return new SpecificDatumReader(clazz.newInstance().schema());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
- protected Schema getSchema(SpecificRecord t) {
+ @Override
+ protected Schema getSchema(SpecificRecord t, Map<String, String> metadata) {
return t.schema();
}
- protected DatumWriter getWriter(Class<SpecificRecord> clazz) {
+ @Override
+ protected DatumWriter getWriter(Map<String, String> metadata) {
return new SpecificDatumWriter();
}
Modified:
hadoop/common/trunk/src/java/org/apache/hadoop/util/ReflectionUtils.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/util/ReflectionUtils.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/util/ReflectionUtils.java
(original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/util/ReflectionUtils.java
Thu Sep 3 01:36:33 2009
@@ -18,21 +18,27 @@
package org.apache.hadoop.util;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
-import java.io.*;
-import java.lang.management.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
-import org.apache.hadoop.conf.*;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.DeserializerBase;
+import org.apache.hadoop.io.serializer.SerializationBase;
import org.apache.hadoop.io.serializer.SerializationFactory;
-import org.apache.hadoop.io.serializer.Serializer;
+import org.apache.hadoop.io.serializer.SerializerBase;
/**
* General reflection utils
@@ -269,11 +275,12 @@
buffer.outBuffer.reset();
SerializationFactory factory = getFactory(conf);
Class<T> cls = (Class<T>) src.getClass();
- Serializer<T> serializer = factory.getSerializer(cls);
+ Map<String, String> metadata = SerializationBase.getMetadataFromClass(cls);
+ SerializerBase<T> serializer = factory.getSerializer(metadata);
serializer.open(buffer.outBuffer);
serializer.serialize(src);
buffer.moveData();
- Deserializer<T> deserializer = factory.getDeserializer(cls);
+ DeserializerBase<T> deserializer = factory.getDeserializer(metadata);
deserializer.open(buffer.inBuffer);
dst = deserializer.deserialize(dst);
return dst;
Modified:
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/SerializationTestUtil.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/SerializationTestUtil.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/SerializationTestUtil.java
(original)
+++
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/SerializationTestUtil.java
Thu Sep 3 01:36:33 2009
@@ -17,6 +17,8 @@
*/
package org.apache.hadoop.io.serializer;
+import java.util.Map;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
@@ -33,19 +35,33 @@
* @return deserialized item
*/
public static<K> K testSerialization(Configuration conf, K before)
- throws Exception {
-
+ throws Exception {
+ Map<String, String> metadata =
+ SerializationBase.getMetadataFromClass(GenericsUtil.getClass(before));
+ return testSerialization(conf, metadata, before);
+ }
+
+ /**
+ * A utility that tests serialization/deserialization.
+ * @param conf configuration to use, "io.serializations" is read to
+ * determine the serialization
+ * @param metadata the metadata to pass to the serializer/deserializer
+ * @param <K> the class of the item
+ * @param before item to (de)serialize
+ * @return deserialized item
+ */
+ public static <K> K testSerialization(Configuration conf,
+ Map<String, String> metadata, K before) throws Exception {
+
SerializationFactory factory = new SerializationFactory(conf);
- Serializer<K> serializer
- = factory.getSerializer(GenericsUtil.getClass(before));
- Deserializer<K> deserializer
- = factory.getDeserializer(GenericsUtil.getClass(before));
-
+ SerializerBase<K> serializer = factory.getSerializer(metadata);
+ DeserializerBase<K> deserializer = factory.getDeserializer(metadata);
+
DataOutputBuffer out = new DataOutputBuffer();
serializer.open(out);
serializer.serialize(before);
serializer.close();
-
+
DataInputBuffer in = new DataInputBuffer();
in.reset(out.getData(), out.getLength());
deserializer.open(in);
Modified:
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/avro/TestAvroSerialization.java
URL:
http://svn.apache.org/viewvc/hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/avro/TestAvroSerialization.java?rev=810756&r1=810755&r2=810756&view=diff
==============================================================================
---
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/avro/TestAvroSerialization.java
(original)
+++
hadoop/common/trunk/src/test/core/org/apache/hadoop/io/serializer/avro/TestAvroSerialization.java
Thu Sep 3 01:36:33 2009
@@ -18,9 +18,14 @@
package org.apache.hadoop.io.serializer.avro;
+import java.util.HashMap;
+import java.util.Map;
+
import junit.framework.TestCase;
+import org.apache.avro.util.Utf8;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.serializer.SerializationBase;
import org.apache.hadoop.io.serializer.SerializationTestUtil;
public class TestAvroSerialization extends TestCase {
@@ -59,6 +64,16 @@
SerializationTestUtil.testSerialization(conf, before);
assertEquals(before, after);
}
+
+ public void testGeneric() throws Exception {
+ Utf8 before = new Utf8("hadoop");
+ Map<String, String> metadata = new HashMap<String, String>();
+ metadata.put(SerializationBase.SERIALIZATION_KEY,
+ AvroGenericSerialization.class.getName());
+ metadata.put(AvroSerialization.AVRO_SCHEMA_KEY, "\"string\"");
+ Utf8 after = SerializationTestUtil.testSerialization(conf, metadata,
before);
+ assertEquals(before, after);
+ }
public static class InnerRecord {
public int x = 7;
