diff --git a/src/main/java/com/profiler/common/bo/AnnotationBo.java b/src/main/java/com/profiler/common/bo/AnnotationBo.java
new file mode 100644
index 000000000..fde0ce013
--- /dev/null
+++ b/src/main/java/com/profiler/common/bo/AnnotationBo.java
@@ -0,0 +1,164 @@
+package com.profiler.common.bo;
+
+import com.profiler.common.dto.thrift.Annotation;
+import com.profiler.common.util.Buffer;
+
+import java.nio.charset.Charset;
+
+/**
+ *
+ */
+public class AnnotationBo {
+    private static final Charset UTF8 = Charset.forName("UTF-8");
+
+    private static final int VERSION_SIZE = 1;
+    // version 0 = prefix의 사이즈를 int로
+    // version 1 = prefix의 사이즈를 short로
+    // version 2 = prefix의 사이즈를 byte 하면 byte eocnding이 좀 줄지 않나?
+    private byte version = 0;
+    private long spanId;
+    private long timestamp;
+    private String key;
+    private byte[] keyBytes;
+
+    private long duration;
+    private int valueType;
+    private byte[] value;
+
+
+    public AnnotationBo() {
+    }
+
+    public AnnotationBo(Annotation ano) {
+        this.timestamp = ano.getTimestamp();
+        this.key = ano.getKey();
+        this.duration = ano.getDuration();
+        this.valueType = ano.getValueTypeCode();
+        this.value = ano.getValue();
+    }
+
+    public long getSpanId() {
+        return spanId;
+    }
+
+    public void setSpanId(long spanId) {
+        this.spanId = spanId;
+    }
+
+    public int getVersion() {
+        return version & 0xFF;
+    }
+
+    public void setVersion(int version) {
+        if (version < 0 || version > 255) {
+            throw new IllegalArgumentException("out of range (0~255)");
+        }
+        // range 체크
+        this.version = (byte) (version & 0xFF);
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public byte[] getKeyBytes() {
+        if (keyBytes == null) {
+            keyBytes = this.key.getBytes(UTF8);
+        }
+        return keyBytes;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public long getDuration() {
+        return duration;
+    }
+
+    public void setDuration(long duration) {
+        this.duration = duration;
+    }
+
+    public int getValueType() {
+        return valueType;
+    }
+
+    public void setValueType(int valueType) {
+        this.valueType = valueType;
+    }
+
+    public byte[] getValue() {
+        return value;
+    }
+
+    public void setValue(byte[] value) {
+        this.value = value;
+    }
+
+
+    public int writeValue(byte[] buf, int offset) {
+//        long timestamp; // required 8
+//        long duration; // optional 8
+//        String key; // required 4+string.length
+//        int valueTypeCode; // required 4
+//        ByteBuffer value; // optional 4 + buf.length
+        Buffer buffer = new Buffer(buf, offset);
+        buffer.put(this.version);
+        buffer.put(this.timestamp);
+        buffer.put(this.duration);
+        buffer.putPrefixedBytes(getKeyBytes());
+        buffer.put(this.valueType);
+        buffer.putPrefixedBytes(value);
+        return buffer.getOffset();
+    }
+
+    public int getBufferSize() {
+//        long timestamp; // required 8
+//        long duration; // optional 8
+//        String key; // required 4+string.length
+//        int valueTypeCode; // required 4
+//        ByteBuffer value; // optional 4 + buf.length
+        int size = 0;
+        size += 1 + 8 + 8 + 4 + 4 + 4;
+        size += this.getKeyBytes().length;
+        if (this.getValue() != null) {
+            size += this.getValue().length;
+        }
+        return size;
+    }
+
+
+    public int readValue(byte[] buf, int offset) {
+        Buffer buffer = new Buffer(buf, offset);
+        this.version = buffer.readByte();
+        this.timestamp = buffer.readLong();
+        this.duration = buffer.readLong();
+        this.key = buffer.readPrefixedString();
+        this.valueType = buffer.readInt();
+        this.value = buffer.readPrefixedBytes();
+        return buffer.getOffset();
+    }
+
+    @Override
+    public String toString() {
+        return "AnnotationBo{" +
+                "version=" + version +
+                ", spanId=" + spanId +
+                ", timestamp=" + timestamp +
+                ", key='" + key + '\'' +
+                ", keyBytes=" + keyBytes +
+                ", duration=" + duration +
+                ", valueType=" + valueType +
+                ", value=" + value +
+                '}';
+    }
+}
diff --git a/src/main/java/com/profiler/common/bo/SpanBo.java b/src/main/java/com/profiler/common/bo/SpanBo.java
new file mode 100644
index 000000000..efc15079a
--- /dev/null
+++ b/src/main/java/com/profiler/common/bo/SpanBo.java
@@ -0,0 +1,276 @@
+package com.profiler.common.bo;
+
+import com.profiler.common.dto.thrift.Annotation;
+import com.profiler.common.dto.thrift.Span;
+import com.profiler.common.util.Buffer;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ *
+ */
+public class SpanBo {
+
+    private static final Charset UTF8 = Charset.forName("UTF-8");
+
+    private static final int VERSION_SIZE = 1;
+    // version 0 = prefix의 사이즈를 int로
+    // version 1 = prefix의 사이즈를 short로
+    // version 2 = prefix의 사이즈를 byte 하면 byte eocnding이 좀 줄지 않나?
+    private byte version = 0;
+
+    private String agentId; // required
+
+    //    private static final int TIMESTAMP = 8;
+    private long timestamp; // required
+
+    //    private static final int MOSTTRACEID = 8;
+    private long mostTraceId; // required
+
+    //    private static final int LEASTTRACEID = 8;
+    private long leastTraceId; // required
+
+    private String name; // required
+    private String serviceName; // required
+
+    //    private static final int SPANID = 8;
+    private long spanId; // required
+
+    private static final int PARENTSPANID = 8;
+    private long parentSpanId; // optional
+
+    private static final int FLAG = 4;
+    private int flag; // optional
+    //    private List<Annotation> annotations; // required
+
+    private String endPoint; // required
+
+    private static final int TERMINAL = 1;
+    private boolean terminal; // required
+
+    private List<AnnotationBo> annotationBoList;
+
+    public SpanBo(Span span) {
+        this.agentId = span.getAgentId();
+        this.timestamp = span.getTimestamp();
+        this.mostTraceId = span.getMostTraceId();
+        this.leastTraceId = span.getLeastTraceId();
+        this.name = span.getName();
+        this.serviceName = span.getServiceName();
+        this.spanId = span.getSpanId();
+        this.parentSpanId = span.getParentSpanId();
+        this.endPoint = span.getEndPoint();
+        this.flag = span.getFlag();
+        this.terminal = span.isTerminal();
+        setAnnotationList(span.getAnnotations());
+    }
+
+    public SpanBo(long mostTraceId, long timestamp, long leastTraceId, long spanId) {
+        this.mostTraceId = mostTraceId;
+        this.timestamp = timestamp;
+        this.leastTraceId = leastTraceId;
+        this.spanId = spanId;
+    }
+
+    public SpanBo() {
+    }
+
+    public int getVersion() {
+        return version & 0xFF;
+    }
+
+    public void setVersion(int version) {
+        if (version < 0 || version > 255) {
+            throw new IllegalArgumentException("out of range (0~255)");
+        }
+        // range 체크
+        this.version = (byte) (version & 0xFF);
+    }
+
+    public String getAgentId() {
+        return agentId;
+    }
+
+    public void setAgentId(String agentId) {
+        this.agentId = agentId;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public long getMostTraceId() {
+        return mostTraceId;
+    }
+
+    public void setMostTraceId(long mostTraceId) {
+        this.mostTraceId = mostTraceId;
+    }
+
+    public long getLeastTraceId() {
+        return leastTraceId;
+    }
+
+    public void setLeastTraceId(long leastTraceId) {
+        this.leastTraceId = leastTraceId;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getServiceName() {
+        return serviceName;
+    }
+
+    public void setServiceName(String serviceName) {
+        this.serviceName = serviceName;
+    }
+
+    public long getSpanId() {
+        return spanId;
+    }
+
+    public void setSpanID(long spanId) {
+        this.spanId = spanId;
+    }
+
+    public long getParentSpanId() {
+        return parentSpanId;
+    }
+
+    public void setParentSpanId(long parentSpanId) {
+        this.parentSpanId = parentSpanId;
+    }
+
+    public int getFlag() {
+        return flag;
+    }
+
+    public void setFlag(int flag) {
+        this.flag = flag;
+    }
+
+    public String getEndPoint() {
+        return endPoint;
+    }
+
+    public void setEndPoint(String endPoint) {
+        this.endPoint = endPoint;
+    }
+
+    public boolean isTerminal() {
+        return terminal;
+    }
+
+    public void setTerminal(boolean terminal) {
+        this.terminal = terminal;
+    }
+
+    public List<AnnotationBo> getAnnotationBoList() {
+        if (annotationBoList == null) {
+            return Collections.emptyList();
+        }
+        return annotationBoList;
+    }
+
+
+    public void setAnnotationList(List<Annotation> anoList) {
+        List<AnnotationBo> boList = new ArrayList<AnnotationBo>(anoList.size());
+        for (Annotation ano : anoList) {
+            boList.add(new AnnotationBo(ano));
+        }
+        this.annotationBoList = boList;
+    }
+
+    public void setAnnotationBoList(List<AnnotationBo> anoList) {
+//        List<AnnotationBo> boList = new ArrayList<AnnotationBo>(anoList.size());
+//        for(Annotation ano : anoList) {
+//            boList.add(new AnnotationBo(ano));
+//        }
+//        this.annotationBoList = boList;
+        if (anoList == null) {
+            this.annotationBoList = Collections.emptyList();
+        } else {
+            this.annotationBoList = anoList;
+        }
+    }
+
+    private int getBufferLength(int a, int b, int c, int d) {
+        int size = a + b + c + d;
+        size = size + (4 * 4) + VERSION_SIZE; // chunk
+//        size = size + TIMESTAMP + MOSTTRACEID + LEASTTRACEID + SPANID + PARENTSPANID + FLAG + TERMINAL;
+        size = size + PARENTSPANID + FLAG + TERMINAL;
+        return size;
+    }
+
+    public byte[] writeValue() {
+
+        byte[] agentIDBytes = agentId.getBytes(UTF8);
+        byte[] nameBytes = name.getBytes(UTF8);
+        byte[] serviceNameBytes = serviceName.getBytes(UTF8);
+        byte[] endPointBytes = endPoint.getBytes(UTF8);
+        int bufferLength = getBufferLength(agentIDBytes.length, nameBytes.length, serviceNameBytes.length, endPointBytes.length);
+
+        Buffer buffer = new Buffer(bufferLength);
+        buffer.put(version);
+        buffer.putPrefixedBytes(agentIDBytes);
+//        buffer.put(timestamp);
+//        buffer.put(mostTraceID);
+//        buffer.put(leastTraceID);
+        buffer.putPrefixedBytes(nameBytes);
+        buffer.putPrefixedBytes(serviceNameBytes);
+//        buffer.put(spanID);
+        buffer.put(parentSpanId);
+        buffer.put(flag);
+        buffer.putPrefixedBytes(endPointBytes);
+        buffer.put(terminal);
+        return buffer.getBuffer();
+    }
+
+
+    public int readValue(byte[] bytes, int offset) {
+        Buffer buffer = new Buffer(bytes, offset);
+        this.version = buffer.readByte();
+        this.agentId = buffer.readPrefixedString();
+//        this.timestamp = buffer.readLong();
+//        this.mostTraceID = buffer.readLong();
+//        this.leastTraceID = buffer.readLong();
+        this.name = buffer.readPrefixedString();
+        this.serviceName = buffer.readPrefixedString();
+//        this.spanID = buffer.readLong();
+        this.parentSpanId = buffer.readLong();
+        this.flag = buffer.readInt();
+        this.endPoint = buffer.readPrefixedString();
+        this.terminal = buffer.readBoolean();
+        return buffer.getOffset();
+    }
+
+    @Override
+    public String toString() {
+        return "SpanBo{" +
+                "agentId='" + agentId + '\'' +
+                ", timestamp=" + timestamp +
+                ", mostTraceId=" + mostTraceId +
+                ", leastTraceId=" + leastTraceId +
+                ", name='" + name + '\'' +
+                ", serviceName='" + serviceName + '\'' +
+                ", spanID=" + spanId +
+                ", parentSpanId=" + parentSpanId +
+                ", flag=" + flag +
+                ", endPoint='" + endPoint + '\'' +
+                ", terminal=" + terminal +
+                '}';
+    }
+}
diff --git a/src/main/java/com/profiler/common/hbase/HBaseTables.java b/src/main/java/com/profiler/common/hbase/HBaseTables.java
index 5b647bf38..532cc5fa8 100644
--- a/src/main/java/com/profiler/common/hbase/HBaseTables.java
+++ b/src/main/java/com/profiler/common/hbase/HBaseTables.java
@@ -10,6 +10,7 @@ public class HBaseTables {
 
     public static final String TRACES = "Traces";
     public static final byte[] TRACES_CF_SPAN = Bytes.toBytes("Span");
+    public static final byte[] TRACES_CF_ANNOTATION = Bytes.toBytes("Annotation");
 
     public static final String SERVERS = "Servers";
 }
diff --git a/src/main/java/com/profiler/common/hbase/HbaseOperations2.java b/src/main/java/com/profiler/common/hbase/HbaseOperations2.java
index 569d99fe2..54e4ed23a 100644
--- a/src/main/java/com/profiler/common/hbase/HbaseOperations2.java
+++ b/src/main/java/com/profiler/common/hbase/HbaseOperations2.java
@@ -47,6 +47,8 @@ public interface HbaseOperations2 extends HbaseOperations {
      */
     <T> T get(String tableName, final byte[] rowName, final byte[] familyName, final byte[] qualifier, final RowMapper<T> mapper);
 
+    <T> T get(String tableName, final Get get, final RowMapper<T> mapper);
+
     <T> List<T> get(String tableName, final List<Get> get, final RowMapper<T> mapper);
 
 
diff --git a/src/main/java/com/profiler/common/hbase/HbaseTemplate2.java b/src/main/java/com/profiler/common/hbase/HbaseTemplate2.java
index 3d408d97f..8e81f24f6 100644
--- a/src/main/java/com/profiler/common/hbase/HbaseTemplate2.java
+++ b/src/main/java/com/profiler/common/hbase/HbaseTemplate2.java
@@ -44,11 +44,22 @@ public class HbaseTemplate2 extends HbaseTemplate implements HbaseOperations2 {
     }
 
     @Override
-    public <T> List<T> get(String tableName, final List<Get> get, final RowMapper<T> mapper) {
+    public <T> T get(String tableName, final Get get, final RowMapper<T> mapper) {
+        return execute(tableName, new TableCallback<T>() {
+            @Override
+            public T doInTable(HTable htable) throws Throwable {
+                Result result = htable.get(get);
+                return mapper.mapRow(result, 0);
+            }
+        });
+    }
+
+    @Override
+    public <T> List<T> get(String tableName, final List<Get> gets, final RowMapper<T> mapper) {
         return execute(tableName, new TableCallback<List<T>>() {
             @Override
             public List<T> doInTable(HTable htable) throws Throwable {
-                Result[] result = htable.get(get);
+                Result[] result = htable.get(gets);
                 List<T> list = new ArrayList<T>(result.length);
                 for (int i = 0; i < result.length; i++) {
                     T t = mapper.mapRow(result[i], i);
diff --git a/src/main/java/com/profiler/common/util/AnnotationTranscoder.java b/src/main/java/com/profiler/common/util/AnnotationTranscoder.java
index 17ebf8ba8..cd760a968 100644
--- a/src/main/java/com/profiler/common/util/AnnotationTranscoder.java
+++ b/src/main/java/com/profiler/common/util/AnnotationTranscoder.java
@@ -10,184 +10,184 @@ import java.util.Date;
 
 public class AnnotationTranscoder {
 
-	private static final String DEFAULT_CHARSET = "UTF-8";
+    private static final String DEFAULT_CHARSET = "UTF-8";
 
-	static final int SERIALIZED = 1;
+    static final int SERIALIZED = 1;
 
-	// Special flags for specially handled types.
-	protected static final int SPECIAL_MASK = 0xff00;
-	static final int SPECIAL_BOOLEAN = (1 << 8);
-	static final int SPECIAL_INT = (2 << 8);
-	static final int SPECIAL_LONG = (3 << 8);
-	static final int SPECIAL_DATE = (4 << 8);
-	static final int SPECIAL_BYTE = (5 << 8);
-	static final int SPECIAL_FLOAT = (6 << 8);
-	static final int SPECIAL_DOUBLE = (7 << 8);
-	static final int SPECIAL_BYTEARRAY = (8 << 8);
+    // Special flags for specially handled types.
+    protected static final int SPECIAL_MASK = 0xff00;
+    static final int SPECIAL_BOOLEAN = (1 << 8);
+    static final int SPECIAL_INT = (2 << 8);
+    static final int SPECIAL_LONG = (3 << 8);
+    static final int SPECIAL_DATE = (4 << 8);
+    static final int SPECIAL_BYTE = (5 << 8);
+    static final int SPECIAL_FLOAT = (6 << 8);
+    static final int SPECIAL_DOUBLE = (7 << 8);
+    static final int SPECIAL_BYTEARRAY = (8 << 8);
 
-	protected final TranscoderUtils tu = new TranscoderUtils(true);
+    protected final TranscoderUtils tu = new TranscoderUtils(true);
 
-	public static final class Encoded {
+    public static final class Encoded {
 
-		private final int valueType;
-		private final byte[] bytes;
+        private final int valueType;
+        private final byte[] bytes;
 
-		public Encoded(int valueType, byte[] bytes) {
-			this.valueType = valueType;
-			this.bytes = bytes;
-		}
+        public Encoded(int valueType, byte[] bytes) {
+            this.valueType = valueType;
+            this.bytes = bytes;
+        }
 
-		public int getValueType() {
-			return valueType;
-		}
+        public int getValueType() {
+            return valueType;
+        }
 
-		public byte[] getBytes() {
-			return bytes;
-		}
-	}
+        public byte[] getBytes() {
+            return bytes;
+        }
+    }
 
-	public Object decode(int dataType, byte[] data) {
-		Object rv = null;
+    public Object decode(int dataType, byte[] data) {
+        Object rv = null;
 
-		int flags = dataType & SPECIAL_MASK;
+        int flags = dataType & SPECIAL_MASK;
 
-		if ((dataType & SERIALIZED) != 0 && data != null) {
-			rv = deserialize(data);
-		} else if (flags != 0 && data != null) {
-			switch (flags) {
-			case SPECIAL_BOOLEAN:
-				rv = Boolean.valueOf(tu.decodeBoolean(data));
-				break;
-			case SPECIAL_INT:
-				rv = new Integer(tu.decodeInt(data));
-				break;
-			case SPECIAL_LONG:
-				rv = new Long(tu.decodeLong(data));
-				break;
-			case SPECIAL_DATE:
-				rv = new Date(tu.decodeLong(data));
-				break;
-			case SPECIAL_BYTE:
-				rv = new Byte(tu.decodeByte(data));
-				break;
-			case SPECIAL_FLOAT:
-				rv = new Float(Float.intBitsToFloat(tu.decodeInt(data)));
-				break;
-			case SPECIAL_DOUBLE:
-				rv = new Double(Double.longBitsToDouble(tu.decodeLong(data)));
-				break;
-			case SPECIAL_BYTEARRAY:
-				rv = data;
-				break;
-			default:
+        if ((dataType & SERIALIZED) != 0 && data != null) {
+            rv = deserialize(data);
+        } else if (flags != 0 && data != null) {
+            switch (flags) {
+                case SPECIAL_BOOLEAN:
+                    rv = Boolean.valueOf(tu.decodeBoolean(data));
+                    break;
+                case SPECIAL_INT:
+                    rv = new Integer(tu.decodeInt(data));
+                    break;
+                case SPECIAL_LONG:
+                    rv = new Long(tu.decodeLong(data));
+                    break;
+                case SPECIAL_DATE:
+                    rv = new Date(tu.decodeLong(data));
+                    break;
+                case SPECIAL_BYTE:
+                    rv = new Byte(tu.decodeByte(data));
+                    break;
+                case SPECIAL_FLOAT:
+                    rv = new Float(Float.intBitsToFloat(tu.decodeInt(data)));
+                    break;
+                case SPECIAL_DOUBLE:
+                    rv = new Double(Double.longBitsToDouble(tu.decodeLong(data)));
+                    break;
+                case SPECIAL_BYTEARRAY:
+                    rv = data;
+                    break;
+                default:
 //				LOG.warn("Undecodeable with flags %x", flags);
-			}
-		} else {
-			rv = decodeString(data);
-		}
-		return rv;
-	}
+            }
+        } else {
+            rv = decodeString(data);
+        }
+        return rv;
+    }
 
-	public Encoded encode(Object o) {
-		byte[] b = null;
-		int flags = 0;
-		if (o instanceof String) {
-			b = encodeString((String) o);
-		} else if (o instanceof Long) {
-			b = tu.encodeLong((Long) o);
-			flags |= SPECIAL_LONG;
-		} else if (o instanceof Integer) {
-			b = tu.encodeInt((Integer) o);
-			flags |= SPECIAL_INT;
-		} else if (o instanceof Boolean) {
-			b = tu.encodeBoolean((Boolean) o);
-			flags |= SPECIAL_BOOLEAN;
-		} else if (o instanceof Date) {
-			b = tu.encodeLong(((Date) o).getTime());
-			flags |= SPECIAL_DATE;
-		} else if (o instanceof Byte) {
-			b = tu.encodeByte((Byte) o);
-			flags |= SPECIAL_BYTE;
-		} else if (o instanceof Float) {
-			b = tu.encodeInt(Float.floatToRawIntBits((Float) o));
-			flags |= SPECIAL_FLOAT;
-		} else if (o instanceof Double) {
-			b = tu.encodeLong(Double.doubleToRawLongBits((Double) o));
-			flags |= SPECIAL_DOUBLE;
-		} else if (o instanceof byte[]) {
-			b = (byte[]) o;
-			flags |= SPECIAL_BYTEARRAY;
-		} else {
-			b = serialize(o);
-			flags |= SERIALIZED;
-		}
+    public Encoded encode(Object o) {
+        byte[] b = null;
+        int flags = 0;
+        if (o instanceof String) {
+            b = encodeString((String) o);
+        } else if (o instanceof Long) {
+            b = tu.encodeLong((Long) o);
+            flags |= SPECIAL_LONG;
+        } else if (o instanceof Integer) {
+            b = tu.encodeInt((Integer) o);
+            flags |= SPECIAL_INT;
+        } else if (o instanceof Boolean) {
+            b = tu.encodeBoolean((Boolean) o);
+            flags |= SPECIAL_BOOLEAN;
+        } else if (o instanceof Date) {
+            b = tu.encodeLong(((Date) o).getTime());
+            flags |= SPECIAL_DATE;
+        } else if (o instanceof Byte) {
+            b = tu.encodeByte((Byte) o);
+            flags |= SPECIAL_BYTE;
+        } else if (o instanceof Float) {
+            b = tu.encodeInt(Float.floatToRawIntBits((Float) o));
+            flags |= SPECIAL_FLOAT;
+        } else if (o instanceof Double) {
+            b = tu.encodeLong(Double.doubleToRawLongBits((Double) o));
+            flags |= SPECIAL_DOUBLE;
+        } else if (o instanceof byte[]) {
+            b = (byte[]) o;
+            flags |= SPECIAL_BYTEARRAY;
+        } else {
+            b = serialize(o);
+            flags |= SERIALIZED;
+        }
 
-		assert b != null;
+        assert b != null;
 
-		return new Encoded(flags, b);
-	}
+        return new Encoded(flags, b);
+    }
 
-	protected byte[] serialize(Object o) {
-		if (o == null) {
-			throw new NullPointerException("Can't serialize null");
-		}
-		byte[] rv = null;
-		try {
-			ByteArrayOutputStream bos = new ByteArrayOutputStream();
-			ObjectOutputStream os = new ObjectOutputStream(bos);
-			os.writeObject(o);
-			os.close();
-			bos.close();
-			rv = bos.toByteArray();
-		} catch (IOException e) {
-			throw new IllegalArgumentException("Non-serializable object, cause=" + e.getMessage(), e);
-		}
-		return rv;
-	}
+    protected byte[] serialize(Object o) {
+        if (o == null) {
+            throw new NullPointerException("Can't serialize null");
+        }
+        byte[] rv = null;
+        try {
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            ObjectOutputStream os = new ObjectOutputStream(bos);
+            os.writeObject(o);
+            os.close();
+            bos.close();
+            rv = bos.toByteArray();
+        } catch (IOException e) {
+            throw new IllegalArgumentException("Non-serializable object, cause=" + e.getMessage(), e);
+        }
+        return rv;
+    }
 
-	protected Object deserialize(byte[] in) {
-		Object rv = null;
-		try {
-			if (in != null) {
-				ByteArrayInputStream bis = new ByteArrayInputStream(in);
-				ObjectInputStream is = new ObjectInputStream(bis);
-				rv = is.readObject();
-				is.close();
-				bis.close();
-			}
-		} catch (IOException e) {
+    protected Object deserialize(byte[] in) {
+        Object rv = null;
+        try {
+            if (in != null) {
+                ByteArrayInputStream bis = new ByteArrayInputStream(in);
+                ObjectInputStream is = new ObjectInputStream(bis);
+                rv = is.readObject();
+                is.close();
+                bis.close();
+            }
+        } catch (IOException e) {
 //			LOG.error(e.getMessage(), e);
-		} catch (ClassNotFoundException e) {
+        } catch (ClassNotFoundException e) {
 //			LOG.error(e.getMessage(), e);
-		}
-		return rv;
-	}
+        }
+        return rv;
+    }
 
-	/**
-	 * Decode the string with the current character set.
-	 */
-	protected String decodeString(byte[] data) {
-		String rv = null;
-		try {
-			if (data != null) {
-				rv = new String(data, DEFAULT_CHARSET);
-			}
-		} catch (UnsupportedEncodingException e) {
-			throw new RuntimeException(e);
-		}
-		return rv;
-	}
+    /**
+     * Decode the string with the current character set.
+     */
+    protected String decodeString(byte[] data) {
+        String rv = null;
+        try {
+            if (data != null) {
+                rv = new String(data, DEFAULT_CHARSET);
+            }
+        } catch (UnsupportedEncodingException e) {
+            throw new RuntimeException(e);
+        }
+        return rv;
+    }
 
-	/**
-	 * Encode a string into the current character set.
-	 */
-	protected byte[] encodeString(String in) {
-		byte[] rv = null;
-		try {
-			rv = in.getBytes(DEFAULT_CHARSET);
-		} catch (UnsupportedEncodingException e) {
-			throw new RuntimeException(e);
-		}
-		return rv;
-	}
+    /**
+     * Encode a string into the current character set.
+     */
+    protected byte[] encodeString(String in) {
+        byte[] rv = null;
+        try {
+            rv = in.getBytes(DEFAULT_CHARSET);
+        } catch (UnsupportedEncodingException e) {
+            throw new RuntimeException(e);
+        }
+        return rv;
+    }
 }
diff --git a/src/main/java/com/profiler/common/util/Buffer.java b/src/main/java/com/profiler/common/util/Buffer.java
new file mode 100644
index 000000000..041f9f5b0
--- /dev/null
+++ b/src/main/java/com/profiler/common/util/Buffer.java
@@ -0,0 +1,138 @@
+package com.profiler.common.util;
+
+import java.nio.charset.Charset;
+
+/**
+ *
+ */
+public class Buffer {
+
+    public static final int BOOLEAN_FALSE = 0;
+    public static final int BOOLEAN_TRUE = 1;
+
+    public static byte[] EMPTY = new byte[0];
+
+    private static final Charset UTF8 = Charset.forName("UTF-8");
+
+    private byte[] buffer;
+    private int offset;
+
+    public Buffer(int size) {
+        this.buffer = new byte[size];
+        this.offset = 0;
+    }
+
+    public Buffer(byte[] buffer) {
+        if (buffer == null) {
+            throw new NullPointerException("buffer must not be null");
+        }
+        this.buffer = buffer;
+        this.offset = 0;
+    }
+
+    public Buffer(byte[] buffer, int offset) {
+        if (buffer == null) {
+            throw new NullPointerException("buffer must not be null");
+        }
+        this.buffer = buffer;
+        this.offset = offset;
+    }
+
+
+    public void putPrefixedBytes(byte[] bytes) {
+        if (bytes == null) {
+            put(0);
+        } else {
+            put(bytes.length);
+            put(bytes);
+        }
+    }
+
+    public byte readByte() {
+        return this.buffer[offset++];
+    }
+
+    public boolean readBoolean() {
+        byte b = readByte();
+        if (b == BOOLEAN_FALSE) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    public int readInt() {
+        int i = BytesUtils.bytesToInt(buffer, offset);
+        this.offset = this.offset + 4;
+        return i;
+    }
+
+    public long readLong() {
+        long l = BytesUtils.bytesToLong(buffer, offset);
+        this.offset = this.offset + 8;
+        return l;
+    }
+
+    public byte[] readPrefixedBytes() {
+        int size = readInt();
+        if (size == 0) {
+            return EMPTY;
+        }
+        byte[] b = new byte[size];
+        System.arraycopy(buffer, offset, b, 0, size);
+        this.offset = offset + size;
+        return b;
+    }
+
+    public String readPrefixedString() {
+        int size = readInt();
+        if (size == 0) {
+            return "";
+        }
+        String s = new String(buffer, offset, size, UTF8);
+        this.offset = offset + size;
+        return s;
+    }
+
+
+    public void put(byte v) {
+        this.buffer[offset++] = v;
+    }
+
+    public void put(boolean v) {
+        if (v) {
+            this.buffer[offset++] = BOOLEAN_TRUE;
+        } else {
+            this.buffer[offset++] = BOOLEAN_FALSE;
+        }
+    }
+
+    public void put(int v) {
+        BytesUtils.writeInt(v, buffer, offset);
+        this.offset = offset + 4;
+    }
+
+    public void put(long v) {
+        BytesUtils.writeLong(v, buffer, offset);
+        this.offset = offset + 8;
+    }
+
+    public void put(byte[] v) {
+        if (v == null) {
+            throw new NullPointerException("v must not be null");
+        }
+        System.arraycopy(v, 0, buffer, offset, v.length);
+        this.offset = offset + v.length;
+    }
+
+
+    public byte[] getBuffer() {
+        return this.buffer;
+    }
+
+    public int getOffset() {
+        return offset;
+    }
+
+
+}
diff --git a/src/main/java/com/profiler/common/util/BytesUtils.java b/src/main/java/com/profiler/common/util/BytesUtils.java
index 74c625678..9863a6be4 100644
--- a/src/main/java/com/profiler/common/util/BytesUtils.java
+++ b/src/main/java/com/profiler/common/util/BytesUtils.java
@@ -46,6 +46,22 @@ public class BytesUtils {
         return rv;
     }
 
+    public static int bytesToInt(byte[] buf, int offset) {
+        if (buf == null) {
+            throw new NullPointerException("buf must not be null");
+        }
+        if (buf.length < offset + 4) {
+            throw new IllegalArgumentException("buf.length is too small. buf.length:" + buf.length + " offset:" + offset + 4);
+        }
+
+        int v = ((buf[offset] & 0xff) << 24)
+                | ((buf[offset + 1] & 0xff) << 16)
+                | ((buf[offset + 2] & 0xff) << 8)
+                | ((buf[offset + 3] & 0xff));
+
+        return v;
+    }
+
     public static long bytesToFirstLong(byte[] buf) {
         if (buf == null) {
             throw new NullPointerException("buf must not be null");
@@ -101,6 +117,30 @@ public class BytesUtils {
         buf[offset] = (byte) (value);
     }
 
+    public static void writeShort(short value, byte[] buf, int offset) {
+        if (buf == null) {
+            throw new NullPointerException("buf must not be null");
+        }
+        if (buf.length < offset + 4) {
+            throw new IllegalArgumentException("buf.length is too small. buf.length:" + buf.length + " offset:" + offset + 4);
+        }
+        buf[offset++] = (byte) (value >> 8);
+        buf[offset] = (byte) (value);
+    }
+
+    public static void writeInt(int value, byte[] buf, int offset) {
+        if (buf == null) {
+            throw new NullPointerException("buf must not be null");
+        }
+        if (buf.length < offset + 4) {
+            throw new IllegalArgumentException("buf.length is too small. buf.length:" + buf.length + " offset:" + offset + 4);
+        }
+        buf[offset++] = (byte) (value >> 24);
+        buf[offset++] = (byte) (value >> 16);
+        buf[offset++] = (byte) (value >> 8);
+        buf[offset] = (byte) (value);
+    }
+
     public static void writeFirstLong(long value, byte[] buf) {
         if (buf == null) {
             throw new NullPointerException("buf must not be null");
@@ -146,4 +186,6 @@ public class BytesUtils {
         writeLong(fostfix, buf, preFix.length);
         return buf;
     }
+
+
 }
diff --git a/src/main/java/com/profiler/common/util/SpanUtils.java b/src/main/java/com/profiler/common/util/SpanUtils.java
index 58cbb5998..daf0624e7 100644
--- a/src/main/java/com/profiler/common/util/SpanUtils.java
+++ b/src/main/java/com/profiler/common/util/SpanUtils.java
@@ -1,24 +1,14 @@
 package com.profiler.common.util;
 
-import org.apache.commons.lang.ArrayUtils;
-import org.apache.hadoop.hbase.util.Bytes;
-
 import com.profiler.common.dto.thrift.Span;
 
 public class SpanUtils {
 
-	public static byte[] getTraceIndexRowKey(Span span) {
-		byte[] agentId = Bytes.toBytes(span.getAgentId());
-		byte[] time = Bytes.toBytes(span.getTimestamp());
+    public static byte[] getTraceIndexRowKey(Span span) {
+        return BytesUtils.add(span.getAgentId(), span.getTimestamp());
+    }
 
-		return ArrayUtils.addAll(agentId, time);
-	}
-
-	public static byte[] getTracesRowkey(Span span) {
+    public static byte[] getTraceId(Span span) {
         return BytesUtils.longLongToBytes(span.getMostTraceId(), span.getLeastTraceId());
-	}
-
-	public static byte[] getTraceId(Span span) {
-        return BytesUtils.longLongToBytes(span.getMostTraceId(), span.getLeastTraceId());
-	}
+    }
 }
diff --git a/src/test/java/com/profiler/common/bo/AnnotationBoTest.java b/src/test/java/com/profiler/common/bo/AnnotationBoTest.java
new file mode 100644
index 000000000..63eeb925b
--- /dev/null
+++ b/src/test/java/com/profiler/common/bo/AnnotationBoTest.java
@@ -0,0 +1,38 @@
+package com.profiler.common.bo;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ *
+ */
+public class AnnotationBoTest {
+    @Test
+    public void testGetVersion() throws Exception {
+
+    }
+
+    @Test
+    public void testSetVersion() throws Exception {
+
+    }
+
+    @Test
+    public void testWriteValue() throws Exception {
+        AnnotationBo bo = new AnnotationBo();
+        bo.setKey("test");
+        bo.setValue("value".getBytes("UTF-8"));
+        int bufferSize = bo.getBufferSize();
+        byte[] bytes = new byte[bufferSize];
+        bo.writeValue(bytes, 0);
+
+        AnnotationBo bo2 = new AnnotationBo();
+        bo2.readValue(bytes, 0);
+        Assert.assertEquals(bo.getKey(), bo2.getKey());
+        Assert.assertEquals(bo.getDuration(), bo2.getDuration());
+        Assert.assertEquals(bo.getTimestamp(), bo2.getTimestamp());
+        Assert.assertEquals(bo.getValueType(), bo2.getValueType());
+        Assert.assertArrayEquals(bo.getValue(), bo2.getValue());
+    }
+
+}
diff --git a/src/test/java/com/profiler/common/bo/SpanBoTest.java b/src/test/java/com/profiler/common/bo/SpanBoTest.java
new file mode 100644
index 000000000..09e74bb6f
--- /dev/null
+++ b/src/test/java/com/profiler/common/bo/SpanBoTest.java
@@ -0,0 +1,55 @@
+package com.profiler.common.bo;
+
+
+import com.profiler.common.bo.SpanBo;
+import junit.framework.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ */
+public class SpanBoTest {
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    @Test
+    public void testVersion() {
+        SpanBo spanBo = new SpanBo();
+        check(spanBo, 0);
+        check(spanBo, 254);
+        check(spanBo, 255);
+        try {
+            check(spanBo, 256);
+            Assert.fail();
+        } catch (Exception e) {
+        }
+
+
+    }
+
+    private void check(SpanBo spanBo, int v) {
+        spanBo.setVersion(v);
+        int version = spanBo.getVersion();
+
+        Assert.assertEquals(v, version);
+    }
+
+    @Test
+    public void serialize() {
+        SpanBo spanBo = new SpanBo();
+        spanBo.setAgentId("agent");
+        spanBo.setEndPoint("end");
+        spanBo.setName("name");
+        spanBo.setServiceName("serviceName");
+
+        byte[] bytes = spanBo.writeValue();
+        logger.debug("length:{}", bytes.length);
+
+        SpanBo newSpanBo = new SpanBo();
+        int i = newSpanBo.readValue(bytes, 0);
+        logger.debug("length:{}", i);
+        Assert.assertEquals(bytes.length, i);
+
+    }
+}
diff --git a/src/test/java/com/profiler/common/util/BytesUtilsTest.java b/src/test/java/com/profiler/common/util/BytesUtilsTest.java
index b7b5b229f..a2a02321b 100644
--- a/src/test/java/com/profiler/common/util/BytesUtilsTest.java
+++ b/src/test/java/com/profiler/common/util/BytesUtilsTest.java
@@ -18,6 +18,21 @@ public class BytesUtilsTest {
         test(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
     }
 
+    @Test
+    public void testInt() {
+        int i = Integer.MAX_VALUE - 5;
+        checkInt(i);
+        checkInt(23464);
+    }
+
+    private void checkInt(int i) {
+        byte[] bytes = Bytes.toBytes(i);
+        int i2 = BytesUtils.bytesToInt(bytes, 0);
+        Assert.assertEquals(i, i2);
+        int i3 = Bytes.toInt(bytes);
+        Assert.assertEquals(i, i3);
+    }
+
     private void test(long most, long least) {
         byte[] bytes1 = Bytes.toBytes(most);
         byte[] bytes2 = Bytes.toBytes(least);