1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mapreduce;
19
20 import java.io.DataInputStream;
21 import java.io.DataOutputStream;
22 import java.io.IOException;
23 import java.io.InputStream;
24 import java.io.OutputStream;
25
26 import org.apache.hadoop.hbase.KeyValue;
27 import org.apache.hadoop.hbase.KeyValueUtil;
28 import org.apache.hadoop.io.serializer.Deserializer;
29 import org.apache.hadoop.io.serializer.Serialization;
30 import org.apache.hadoop.io.serializer.Serializer;
31
32 public class KeyValueSerialization implements Serialization<KeyValue> {
33 @Override
34 public boolean accept(Class<?> c) {
35 return KeyValue.class.isAssignableFrom(c);
36 }
37
38 @Override
39 public KeyValueDeserializer getDeserializer(Class<KeyValue> t) {
40 return new KeyValueDeserializer();
41 }
42
43 @Override
44 public KeyValueSerializer getSerializer(Class<KeyValue> c) {
45 return new KeyValueSerializer();
46 }
47
48 public static class KeyValueDeserializer implements Deserializer<KeyValue> {
49 private DataInputStream dis;
50
51 @Override
52 public void close() throws IOException {
53 this.dis.close();
54 }
55
56 @Override
57 public KeyValue deserialize(KeyValue ignore) throws IOException {
58
59 return KeyValueUtil.create(this.dis);
60 }
61
62 @Override
63 public void open(InputStream is) throws IOException {
64 this.dis = new DataInputStream(is);
65 }
66 }
67
68 public static class KeyValueSerializer implements Serializer<KeyValue> {
69 private DataOutputStream dos;
70
71 @Override
72 public void close() throws IOException {
73 this.dos.close();
74 }
75
76 @Override
77 public void open(OutputStream os) throws IOException {
78 this.dos = new DataOutputStream(os);
79 }
80
81 @Override
82 public void serialize(KeyValue kv) throws IOException {
83 KeyValueUtil.write(kv, this.dos);
84 }
85 }
86 }