diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/FileScanDescriptor.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/FileScanDescriptor.java
index 298a989..5576fd4 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/FileScanDescriptor.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/FileScanDescriptor.java
@@ -9,6 +9,7 @@
 import org.apache.hadoop.fs.Path;
 
 import edu.uci.ics.genomix.data.serde.ByteSerializerDeserializer;
+import edu.uci.ics.genomix.type.Kmer;
 import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
 import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
 import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
@@ -72,10 +73,7 @@
 			private ByteBuffer outputBuffer;
 			private FrameTupleAppender outputAppender;
 			
-			private byte filter0;
-			private byte filter1;
-			private byte filter2;
-			private byte filter3;
+			private byte []filter = new byte[4];
 
 			@SuppressWarnings("resource")
 			@Override
@@ -86,21 +84,7 @@
 				outputAppender = new FrameTupleAppender(ctx.getFrameSize());
 				outputAppender.reset(outputBuffer, true);
 				
-				filter0 = (byte) 0xC0;
-				filter1 = (byte) 0xFC;
-				filter2 = 0;
-				filter3 = 3;
-
-				int r = byteNum * 8 - 2 * k;
-				r = 8 - r;
-				for (int i = 0; i < r; i++) {
-					filter2 <<= 1;
-					filter2 |= 1;
-				}
-				for(int i = 0; i < r-1 ; i++){
-					filter3 <<= 1;
-				}
-				
+				Kmer.initializeFilter(k, filter);
 				
 				try {// one try with multiple catch?
 					writer.open();
@@ -143,113 +127,6 @@
 				}
 			}
 
-			private byte[] CompressKmer(byte[] array, int start) {
-				// a: 00; c: 01; G: 10; T: 11
-
-				byte[] bytes = new byte[byteNum + 1];
-				bytes[0] = (byte) k;
-
-				byte l = 0;
-				int count = 0;
-				int bcount = 0;
-
-				for (int i = start; i < start+k ; i++) {
-					l <<= 2;
-					switch (array[i]) {
-					case 'A':
-					case 'a':
-						l |= 0;
-						break;
-					case 'C':
-					case 'c':
-						l |= 1;
-						break;
-					case 'G':
-					case 'g':
-						l |= 2;
-						break;
-					case 'T':
-					case 't':
-						l |= 3;
-						break;
-					}
-					count += 2;
-					if (count % 8 == 0 && byteNum != bcount + 1) {
-						bytes[byteNum-bcount] = l;
-						bcount += 1;
-						count = 0;
-						l = 0;
-					}
-				}
-				bytes[1] = l;
-				return bytes;
-			}
-
-			private byte GetBitmap(byte t) {
-				byte r = 0;
-				switch (t) {
-				case 'A':
-				case 'a':
-					r = 1;
-					break;
-				case 'C':
-				case 'c':
-					r = 2;
-					break;
-				case 'G':
-				case 'g':
-					r = 4;
-					break;
-				case 'T':
-				case 't':
-					r = 8;
-					break;
-				}
-				return r;
-			}
-
-			private byte ConvertSymbol(byte t) {
-				byte r = 0;
-				switch (t) {
-				case 'A':
-				case 'a':
-					r = 0;
-					break;
-				case 'C':
-				case 'c':
-					r = 1;
-					break;
-				case 'G':
-				case 'g':
-					r = 2;
-					break;
-				case 'T':
-				case 't':
-					r = 3;
-					break;
-				}
-				return r;
-			}
-
-			void MoveKmer(byte[] bytes, byte c) {
-				int i = byteNum;
-				bytes[i] <<= 2;
-				bytes[i] &= filter1;
-				i -= 1;
-				while (i > 1) {
-					byte f = (byte) (bytes[i] & filter0);
-					f >>= 6;
-					f &= 3;
-					bytes[i + 1] |= f;
-					bytes[i] <<= 2;
-					bytes[i] &= filter1;
-					i -= 1;
-				}
-				bytes[2] |= (byte) (bytes[1]&filter3);
-				bytes[1] <<=2;
-				bytes[1] &= filter2;
-				bytes[1] |= ConvertSymbol(c);
-			}
 
 			private void SplitReads(byte[] array) {
 				try {
@@ -260,17 +137,17 @@
 
 					for (int i = 0; i < array.length - k + 1; i++) {
 						if (0 == i) {
-							bytes = CompressKmer(array, i);
+							bytes = Kmer.CompressKmer(k, array, i);
 						} else {
-							MoveKmer(bytes, array[i + k - 1]);
+							Kmer.MoveKmer(k, bytes, array[i+k-1], filter);
 							/*
 							 * l <<= 2; l &= window; l |= ConvertSymbol(array[i
 							 * + k - 1]);
 							 */
-							pre = GetBitmap(array[i - 1]);
+							pre = Kmer.GENE_CODE.getAdjBit(array[i - 1]);
 						}
 						if (i + k != array.length) {
-							next = GetBitmap(array[i + k]);
+							next = Kmer.GENE_CODE.getAdjBit(array[i + k]);
 						}
 
 						r = 0;
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerSequenceWriterFactory.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerSequenceWriterFactory.java
index 6f856de..6928f18 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerSequenceWriterFactory.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerSequenceWriterFactory.java
@@ -8,76 +8,82 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
 import org.apache.hadoop.io.SequenceFile.ValueBytes;
 import org.apache.hadoop.io.SequenceFile.Writer;
+import org.apache.hadoop.mapred.JobConf;
 
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
 import edu.uci.ics.hyracks.hdfs.api.ITupleWriterFactory;
+import edu.uci.ics.hyracks.hdfs.dataflow.ConfFactory;
 
 public class KMerSequenceWriterFactory implements ITupleWriterFactory {
 
 	private static final long serialVersionUID = 1L;
-	private Configuration conf;
-	public KMerSequenceWriterFactory(Configuration conf){
-		this.conf = conf;
+	private ConfFactory confFactory;
+	public KMerSequenceWriterFactory(JobConf conf) throws HyracksDataException{
+			this.confFactory = new ConfFactory(conf);
+	}
+	
+	public class KMerCountValue implements ValueBytes{
+		private ITupleReference tuple;
+		public KMerCountValue(ITupleReference tuple) {
+			this.tuple = tuple;
+		}
+
+		@Override
+		public int getSize() {
+			return tuple.getFieldLength(1) + tuple.getFieldLength(2);
+		}
+
+		@Override
+		public void writeCompressedBytes(DataOutputStream arg0)
+				throws IllegalArgumentException, IOException {
+			for(int i=1; i<3; i++){
+				arg0.write(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
+			}
+		}
+
+		@Override
+		public void writeUncompressedBytes(DataOutputStream arg0)
+				throws IOException {
+			for(int i=1; i<3; i++){
+				arg0.write(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
+			}
+		}
+		
+	}
+	
+	public class TupleWriter implements ITupleWriter{
+		public TupleWriter(ConfFactory cf){
+			this.cf = cf;
+		}
+		ConfFactory cf;
+		Writer writer = null;
+		/**
+		 * assumption is that output never change source! 
+		 */
+		@Override
+		public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
+			try {
+				if (writer == null){
+					writer = SequenceFile.createWriter(cf.getConf(), (FSDataOutputStream) output, BytesWritable.class, BytesWritable.class, CompressionType.NONE, null);
+				}
+				byte[] kmer = tuple.getFieldData(0);
+				int keyStart = tuple.getFieldStart(0);
+				int keyLength = tuple.getFieldLength(0);
+				writer.appendRaw(kmer, keyStart, keyLength, new KMerCountValue(tuple));
+			} catch (IOException e) {
+				throw new HyracksDataException(e);
+			}
+		}
 	}
 	
 	@Override
 	public ITupleWriter getTupleWriter() {
-		return new ITupleWriter(){
-
-			@Override
-			public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
-				try {
-					Writer writer = SequenceFile.createWriter(conf, (FSDataOutputStream) output, BytesWritable.class, BytesWritable.class, null, null);
-					byte[] kmer = tuple.getFieldData(0);
-					int keyStart = tuple.getFieldStart(0);
-					int keyLength = tuple.getFieldLength(0);
-					
-					class KMerCountValue implements ValueBytes{
-						private ITupleReference tuple;
-						public KMerCountValue(ITupleReference tuple) {
-							this.tuple = tuple;
-						}
-
-						@Override
-						public int getSize() {
-							return tuple.getFieldLength(1) + tuple.getFieldLength(2);
-						}
-
-						@Override
-						public void writeCompressedBytes(DataOutputStream arg0)
-								throws IllegalArgumentException, IOException {
-							for(int i=1; i<3; i++){
-								arg0.write(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
-							}
-						}
-
-						@Override
-						public void writeUncompressedBytes(DataOutputStream arg0)
-								throws IOException {
-							for(int i=1; i<3; i++){
-								arg0.write(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
-							}
-						}
-						
-					}
-					for (int i = 0; i < 3; i++) {
-						byte[] data = tuple.getFieldData(0);
-						int start = tuple.getFieldStart(0);
-						int len = tuple.getFieldLength(0);
-						output.write(data, start, len);
-						output.writeChars(" ");
-					}
-					writer.appendRaw(kmer, keyStart, keyLength, new KMerCountValue(tuple));
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
-			
-		};
+		return new TupleWriter(confFactory);
 	}
 
 }
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerTextWriterFactory.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerTextWriterFactory.java
index 1bdaabb..ebec17e 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerTextWriterFactory.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerTextWriterFactory.java
@@ -3,6 +3,10 @@
 import java.io.DataOutput;
 import java.io.IOException;
 
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+
+import edu.uci.ics.genomix.type.Kmer;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.hdfs.api.ITupleWriter;
@@ -14,28 +18,27 @@
 	 * 
 	 */
 	private static final long serialVersionUID = 1L;
+	private final int KMER;
+	public KMerTextWriterFactory(int kmer){
+		KMER = kmer;
+	}
 
+	public class TupleWriter implements ITupleWriter{
+		@Override
+		public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
+			try {
+				Text.writeString(output, Kmer.recoverKmerFrom(KMER,tuple.getFieldData(0),tuple.getFieldStart(0),tuple.getFieldLength(0)));
+				Text.writeString(output,"\t");
+				Text.writeString(output, Kmer.recoverAdjacent(tuple.getFieldData(1)[tuple.getFieldStart(1)]));
+				Text.writeString(output, "\n");
+			} catch (IOException e) {
+				throw new HyracksDataException(e);
+			}
+		}
+	}
 	@Override
 	public ITupleWriter getTupleWriter() {
-		return new ITupleWriter(){
-			byte newLine = "\n".getBytes()[0];
-			@Override
-			public void write(DataOutput output, ITupleReference tuple) throws HyracksDataException {
-				try {
-					for (int i = 0; i < 3; i++) {
-						byte[] data = tuple.getFieldData(0);
-						int start = tuple.getFieldStart(0);
-						int len = tuple.getFieldLength(0);
-						output.write(new String(data,start,len).getBytes());
-						output.writeChar(' ');
-					}
-					output.writeByte(newLine);
-				} catch (IOException e) {
-					throw new HyracksDataException(e);
-				}
-			}
-			
-		};
+		return new TupleWriter();
 	}
 
 }
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerWriterFactory.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerWriterFactory.java
index 13ee200..b77720a 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerWriterFactory.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/KMerWriterFactory.java
@@ -10,6 +10,7 @@
 public class KMerWriterFactory implements ITupleWriterFactory {
 	private static final long serialVersionUID = 1L;
 
+
 	@Override
 	public ITupleWriter getTupleWriter() {
 		return new ITupleWriter() {
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/ReadsKeyValueParserFactory.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/ReadsKeyValueParserFactory.java
index 1876346..6d0ef74 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/ReadsKeyValueParserFactory.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/dataflow/ReadsKeyValueParserFactory.java
@@ -8,6 +8,7 @@
 import org.apache.hadoop.io.Text;
 
 import edu.uci.ics.genomix.data.serde.ByteSerializerDeserializer;
+import edu.uci.ics.genomix.type.Kmer;
 import edu.uci.ics.hyracks.api.comm.IFrameWriter;
 import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
@@ -23,26 +24,23 @@
 
 	private int k;
 	private int byteNum;
-	private byte filter0;
-	private byte filter1;
-	private byte filter2;
-	private byte filter3;
+	private byte[] filter = new byte[4];
 
 	public ReadsKeyValueParserFactory(int k) {
 		this.k = k;
 		byteNum = (byte) Math.ceil((double) k / 4.0);
-		filter0 = (byte) 0xC0;
-		filter1 = (byte) 0xFC;
-		filter2 = 0;
+		filter[0] = (byte) 0xC0;
+		filter[1] = (byte) 0xFC;
+		filter[2] = 0;
 
 		int r = byteNum * 8 - 2 * k;
 		r = 8 - r;
 		for (int i = 0; i < r; i++) {
-			filter2 <<= 1;
-			filter2 |= 1;
+			filter[2] <<= 1;
+			filter[2] |= 1;
 		}
 		for(int i = 0; i < r-1 ; i++){
-			filter3 <<= 1;
+			filter[3] <<= 1;
 		}
 	}
 
@@ -76,115 +74,6 @@
 				FrameUtils.flushFrame(outputBuffer, writer);
 			}
 
-			private byte[] CompressKmer(byte[] array, int start) {
-				// a: 00; c: 01; G: 10; T: 11
-
-				byte[] bytes = new byte[byteNum + 1];
-				bytes[0] = (byte) k;
-
-				byte l = 0;
-				int count = 0;
-				int bcount = 0;
-
-				for (int i = start; i < start+k ; i++) {
-					l <<= 2;
-					switch (array[i]) {
-					case 'A':
-					case 'a':
-						l |= 0;
-						break;
-					case 'C':
-					case 'c':
-						l |= 1;
-						break;
-					case 'G':
-					case 'g':
-						l |= 2;
-						break;
-					case 'T':
-					case 't':
-						l |= 3;
-						break;
-					}
-					count += 2;
-					if (count % 8 == 0 && byteNum != bcount + 1) {
-						bytes[byteNum-bcount] = l;
-						bcount += 1;
-						count = 0;
-						l = 0;
-					}
-				}
-				bytes[1] = l;
-				return bytes;
-			}
-
-			private byte GetBitmap(byte t) {
-				byte r = 0;
-				switch (t) {
-				case 'A':
-				case 'a':
-					r = 1;
-					break;
-				case 'C':
-				case 'c':
-					r = 2;
-					break;
-				case 'G':
-				case 'g':
-					r = 4;
-					break;
-				case 'T':
-				case 't':
-					r = 8;
-					break;
-				}
-				return r;
-			}
-
-			private byte ConvertSymbol(byte t) {
-				byte r = 0;
-				switch (t) {
-				case 'A':
-				case 'a':
-					r = 0;
-					break;
-				case 'C':
-				case 'c':
-					r = 1;
-					break;
-				case 'G':
-				case 'g':
-					r = 2;
-					break;
-				case 'T':
-				case 't':
-					r = 3;
-					break;
-				}
-				return r;
-			}
-
-			void MoveKmer(byte[] bytes, byte c) {
-				int i = byteNum;
-				bytes[i] <<= 2;
-				bytes[i] &= filter1;
-				i -= 1;
-				while (i > 1) {
-					byte f = (byte) (bytes[i] & filter0);
-					f >>= 6;
-					f &= 3;
-					bytes[i + 1] |= f;
-					bytes[i] <<= 2;
-					bytes[i] &= filter1;
-					i -= 1;
-				}
-				bytes[2] |= (byte) (bytes[1]&filter3);
-				bytes[1] <<=2;
-				bytes[1] &= filter2;
-				bytes[1] |= ConvertSymbol(c);
-			}
-
-
 			private void SplitReads(byte[] array, IFrameWriter writer) {
 				try {
 					byte[] bytes = null;
@@ -194,17 +83,17 @@
 
 					for (int i = 0; i < array.length - k + 1; i++) {
 						if (0 == i) {
-							bytes = CompressKmer(array, i);
+							bytes = Kmer.CompressKmer(k,array, i);
 						} else {
-							MoveKmer(bytes, array[i + k - 1]);
+							Kmer.MoveKmer(k,bytes, array[i + k - 1], filter);
 							/*
 							 * l <<= 2; l &= window; l |= ConvertSymbol(array[i
 							 * + k - 1]);
 							 */
-							pre = GetBitmap(array[i - 1]);
+							pre = Kmer.GENE_CODE.getAdjBit(array[i - 1]);
 						}
 						if (i + k != array.length) {
-							next = GetBitmap(array[i + k]);
+							next = Kmer.GENE_CODE.getAdjBit(array[i + k]);
 						}
 
 						r = 0;
@@ -212,16 +101,11 @@
 						r <<= 4;
 						r |= next;
 
-						/*
-						 * System.out.print(l); System.out.print(' ');
-						 * System.out.print(r); System.out.println();
-						 */
-
 						tupleBuilder.reset();
 
 						// tupleBuilder.addField(Integer64SerializerDeserializer.INSTANCE,
 						// l);
-						tupleBuilder.addField(bytes, 0, byteNum + 1); // ? why +1 
+						tupleBuilder.addField(bytes, 0, byteNum + 1); 
 						tupleBuilder.addField(
 								ByteSerializerDeserializer.INSTANCE, r);
 
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/driver/Driver.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/driver/Driver.java
index 3c4f331..b2ffb41 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/driver/Driver.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/driver/Driver.java
@@ -7,6 +7,8 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 import org.apache.hadoop.util.GenericOptionsParser;
@@ -62,13 +64,13 @@
 		/** add hadoop configurations */
 		URL hadoopCore = job.getClass().getClassLoader()
 				.getResource("core-site.xml");
-		job.getConfiguration().addResource(hadoopCore);
+		job.addResource(hadoopCore);
 		URL hadoopMapRed = job.getClass().getClassLoader()
 				.getResource("mapred-site.xml");
-		job.getConfiguration().addResource(hadoopMapRed);
+		job.addResource(hadoopMapRed);
 		URL hadoopHdfs = job.getClass().getClassLoader()
 				.getResource("hdfs-site.xml");
-		job.getConfiguration().addResource(hadoopHdfs);
+		job.addResource(hadoopHdfs);
 
 		LOG.info("job started");
 		long start = System.currentTimeMillis();
@@ -110,6 +112,7 @@
 			JobSpecification createJob = jobGen.generateJob();
 			execute(createJob);
 		} catch (Exception e) {
+			e.printStackTrace();
 			throw e;
 		}
 	}
@@ -126,7 +129,7 @@
 
 	public static void main(String[] args) throws Exception {
 		GenomixJob job = new GenomixJob();
-		String[] otherArgs = new GenericOptionsParser(job.getConfiguration(),
+		String[] otherArgs = new GenericOptionsParser(job,
 				args).getRemainingArgs();
 		if (otherArgs.length < 4) {
 			System.err.println("Need <serverIP> <port> <input> <output>");
@@ -134,12 +137,12 @@
 		}
 		String ipAddress = otherArgs[0];
 		int port = Integer.parseInt(otherArgs[1]);
-		int numOfDuplicate = job.getConfiguration().getInt(
+		int numOfDuplicate = job.getInt(
 				CPARTITION_PER_MACHINE, 2);
-		boolean bProfiling = job.getConfiguration().getBoolean(IS_PROFILING,
+		boolean bProfiling = job.getBoolean(IS_PROFILING,
 				true);
-		FileInputFormat.setInputPaths(job, otherArgs[2]);
-		FileOutputFormat.setOutputPath(job, new Path(otherArgs[3]));
+		FileInputFormat.setInputPaths(new Job(job), otherArgs[2]);
+		FileOutputFormat.setOutputPath(new Job(job), new Path(otherArgs[3]));
 		Driver driver = new Driver(ipAddress, port, numOfDuplicate);
 		driver.runJob(job, Plan.BUILD_DEBRUJIN_GRAPH, bProfiling);
 	}
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/GenomixJob.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/GenomixJob.java
index 7d6101b..e065b8e 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/GenomixJob.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/GenomixJob.java
@@ -5,7 +5,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.Job;
 
-public class GenomixJob extends Job {
+public class GenomixJob extends Configuration {
 
 	public static final String JOB_NAME = "genomix";
 
@@ -30,11 +30,11 @@
 	public static final String GROUPBY_HYBRID_HASHLEVEL = "genomix.graph.groupby.hybrid.hashlevel";
 
 	public GenomixJob() throws IOException {
-		super(new Configuration(), JOB_NAME);
+		super(new Configuration());
 	}
 
 	public GenomixJob(Configuration conf) throws IOException {
-		super(conf, JOB_NAME);
+		super(conf);
 	}
 
 	/**
@@ -44,19 +44,19 @@
 	 *            desired frame size
 	 */
 	final public void setKmerLength(int kmerlength) {
-		getConfiguration().setInt(KMER_LENGTH, kmerlength);
+		setInt(KMER_LENGTH, kmerlength);
 	}
 
 	final public void setFrameSize(int frameSize) {
-		getConfiguration().setInt(FRAME_SIZE, frameSize);
+		setInt(FRAME_SIZE, frameSize);
 	}
 
 	final public void setFrameLimit(int frameLimit) {
-		getConfiguration().setInt(FRAME_LIMIT, frameLimit);
+		setInt(FRAME_LIMIT, frameLimit);
 	}
 
 	final public void setTableSize(int tableSize) {
-		getConfiguration().setInt(TABLE_SIZE, tableSize);
+		setInt(TABLE_SIZE, tableSize);
 	}
 
 }
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGen.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGen.java
index 6933541..557da6b 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGen.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGen.java
@@ -15,7 +15,7 @@
 			System.nanoTime()).toString();
 
 	public JobGen(GenomixJob job) {
-		this.conf = job.getConfiguration();
+		this.conf = job;
 		this.genomixJob = job;
 		this.initJobConfiguration();
 	}
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGenBrujinGraph.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGenBrujinGraph.java
index d6c72da..b7e703e 100644
--- a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGenBrujinGraph.java
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/job/JobGenBrujinGraph.java
@@ -19,6 +19,8 @@
 import edu.uci.ics.genomix.dataflow.ReadsKeyValueParserFactory;
 import edu.uci.ics.genomix.dataflow.aggregators.DistributedMergeLmerAggregateFactory;
 import edu.uci.ics.genomix.dataflow.aggregators.MergeKmerAggregateFactory;
+import edu.uci.ics.genomix.job.GenomixJob;
+import edu.uci.ics.genomix.job.JobGen;
 import edu.uci.ics.hyracks.api.client.NodeControllerInfo;
 import edu.uci.ics.hyracks.api.constraints.PartitionConstraintHelper;
 import edu.uci.ics.hyracks.api.dataflow.IConnectorDescriptor;
@@ -56,6 +58,7 @@
 		TEXT,BINARY,
 	}
 
+	JobConf job;
 	private static final Log LOG = LogFactory.getLog(JobGenBrujinGraph.class);
 	private final Map<String, NodeControllerInfo> ncMap;
 	private Scheduler scheduler;
@@ -189,12 +192,13 @@
 	public HDFSReadOperatorDescriptor createHDFSReader(JobSpecification jobSpec)
 			throws HyracksDataException {
 		try {
-			InputSplit[] splits = ((JobConf) conf).getInputFormat().getSplits(
-					(JobConf) conf, ncNodeNames.length);
+			
+			InputSplit[] splits = job.getInputFormat().getSplits(
+					job, ncNodeNames.length);
 
 			String[] readSchedule = scheduler.getLocationConstraints(splits);
 			return new HDFSReadOperatorDescriptor(jobSpec, outputRec,
-					(JobConf) conf, splits, readSchedule,
+					job, splits, readSchedule,
 					new ReadsKeyValueParserFactory(kmers));
 		} catch (Exception e) {
 			throw new HyracksDataException(e);
@@ -230,21 +234,23 @@
 		ITupleWriterFactory writer = null;
 		switch (outputFormat){
 		case TEXT:
-			writer = new KMerTextWriterFactory();
+			writer = new KMerTextWriterFactory(kmers);
 			break;
 		case BINARY: default:
-			writer = new KMerSequenceWriterFactory(conf);
+			writer = new KMerSequenceWriterFactory(job);
 			break;
 		}
 		HDFSWriteOperatorDescriptor writeOperator = new HDFSWriteOperatorDescriptor(
-				jobSpec, (JobConf) conf, writer);
+				jobSpec, job, writer);
 
 		PartitionConstraintHelper.addAbsoluteLocationConstraint(jobSpec,
 				writeOperator, ncNodeNames);
 
 		IConnectorDescriptor printConn = new OneToOneConnectorDescriptor(
 				jobSpec);
-		jobSpec.connect(printConn, crossGrouper, 0, writeOperator, 0);
+//		jobSpec.connect(printConn, crossGrouper, 0, writeOperator, 0);
+		jobSpec.connect(printConn, readOperator, 0, writeOperator, 0);
+//		jobSpec.addRoot(readOperator);
 		jobSpec.addRoot(writeOperator);
 
 		if (groupbyType == GroupbyType.PRECLUSTER) {
@@ -255,6 +261,7 @@
 
 	@Override
 	protected void initJobConfiguration() {
+		
 		kmers = conf.getInt(GenomixJob.KMER_LENGTH, 25);
 		frameLimits = conf.getInt(GenomixJob.FRAME_LIMIT, 4096);
 		tableSize = conf.getInt(GenomixJob.TABLE_SIZE, 10485767);
@@ -276,6 +283,7 @@
 		} else {
 			outputFormat = OutputFormat.TEXT;
 		}
+		job = new JobConf(conf);
 	}
 
 }
diff --git a/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/type/Kmer.java b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/type/Kmer.java
new file mode 100644
index 0000000..6570509
--- /dev/null
+++ b/genomix/genomix-core/src/main/java/edu/uci/ics/genomix/type/Kmer.java
@@ -0,0 +1,173 @@
+package edu.uci.ics.genomix.type;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+
+import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class Kmer implements Writable {
+	
+	public final static byte[] GENE_SYMBOL = {'A','C','G','T'};
+	public final static class GENE_CODE{
+		
+		public static final byte A=0;
+		public static final byte C=1;
+		public static final byte G=2;
+		public static final byte T=3;
+		
+		public static byte getCodeFromSymbol(byte ch){
+			byte r = 0;
+			switch (ch) {
+			case 'A':case 'a':
+				r = A;
+				break;
+			case 'C':case 'c':
+				r = C;
+				break;
+			case 'G':case 'g':
+				r = G;
+				break;
+			case 'T':case 't':
+				r = T;
+				break;
+			}
+			return r;
+		}
+		
+		public static byte getSymbolFromCode(byte code){
+			if (code > 3){
+				return '!';
+			}
+			return GENE_SYMBOL[code];
+		}
+		
+		public static byte getAdjBit(byte t) {
+			byte r = 0;
+			switch (t) {
+			case 'A':case 'a':
+				r = 1 << A;
+				break;
+			case 'C':case 'c':
+				r = 1 << C;
+				break;
+			case 'G':case 'g':
+				r = 1 << G;
+				break;
+			case 'T':case 't':
+				r = 1 << T;
+				break;
+			}
+			return r;
+		}
+	}
+
+	public static final byte LOWBITMASK = 0x03;
+	
+	public static String recoverKmerFrom(int k, byte [] keyData, int keyStart, int keyLength ) {
+		StringBuilder sblder = new StringBuilder();
+
+		int outKmer = 0;
+		for ( int i = keyLength-1; i>=0; i--){
+			byte last = keyData[keyStart + i];
+			for( int j = 0; j < 8; j+=2){
+				byte kmer = (byte) ((last >>j) & LOWBITMASK);
+				sblder.append((char)GENE_CODE.getSymbolFromCode(kmer));
+				if ( ++outKmer > k){
+					break;
+				}
+			}
+			if(outKmer >k){
+				break;
+			}
+		}
+		return sblder.toString();
+	}
+	
+	public static String recoverAdjacent(byte number){
+		int incoming = (number & 0xF0) >> 4;
+		int outgoing = number & 0x0F;
+		return String.valueOf(incoming) + '|' + String.valueOf(outgoing);
+	}
+	
+	@Override
+	public void readFields(DataInput arg0) throws IOException {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public void write(DataOutput arg0) throws IOException {
+		// TODO Auto-generated method stub
+
+	}
+	
+	public static void initializeFilter(int k, byte []filter){
+		filter[0] = (byte) 0xC0;
+		filter[1] = (byte) 0xFC;
+		filter[2] = 0;
+		filter[3] = 3;
+		final int byteNum = (byte) Math.ceil((double) k / 4.0);
+		
+		int r = byteNum * 8 - 2 * k;
+		r = 8 - r;
+		for (int i = 0; i < r; i++) {
+			filter[2] <<= 1;
+			filter[2] |= 1;
+		}
+		for(int i = 0; i < r-1 ; i++){
+			filter[3] <<= 1;
+		}
+	}
+	
+	public static byte[] CompressKmer(int k, byte[] array, int start) {
+		final int byteNum = (byte) Math.ceil((double) k / 4.0);
+		byte[] bytes = new byte[byteNum + 1];
+		bytes[0] = (byte) k;
+
+		byte l = 0;
+		int count = 0;
+		int bcount = 0;
+
+		for (int i = start; i < start+k ; i++) {
+			l = (byte) ((l<<2) & 0xFC);
+			l |= GENE_CODE.getCodeFromSymbol(array[i]);
+			count += 2;
+			if (count % 8 == 0 && byteNum - bcount > 1) {
+				bytes[byteNum-bcount] = l;
+				bcount += 1;
+				count = 0;
+				l = 0;
+			}
+			if (byteNum - bcount <= 1){
+				break;
+			}
+		}
+		bytes[1] = l;
+		return bytes;
+	}
+	
+	public static void MoveKmer(int k, byte[] bytes, byte c, byte filter[]) {
+		int i = (byte) Math.ceil((double) k / 4.0);;
+		bytes[i] <<= 2;
+		bytes[i] &= filter[1];
+		i -= 1;
+		while (i > 1) {
+			byte f = (byte) (bytes[i] & filter[0]);
+			f >>= 6;
+			f &= 3;
+			bytes[i + 1] |= f;
+			bytes[i] <<= 2;
+			bytes[i] &= filter[1];
+			i -= 1;
+		}
+		bytes[2] |= (byte) (bytes[1]&filter[3]);
+		bytes[1] <<=2;
+		bytes[1] &= filter[2];
+		bytes[1] |= GENE_CODE.getCodeFromSymbol(c);
+	}
+
+
+}
diff --git a/genomix/genomix-core/src/test/java/edu/uci/ics/genomix/example/jobrun/JobRunTestCase.java b/genomix/genomix-core/src/test/java/edu/uci/ics/genomix/example/jobrun/JobRunTestCase.java
index 8fd5fab..a6ff826 100644
--- a/genomix/genomix-core/src/test/java/edu/uci/ics/genomix/example/jobrun/JobRunTestCase.java
+++ b/genomix/genomix-core/src/test/java/edu/uci/ics/genomix/example/jobrun/JobRunTestCase.java
@@ -11,6 +11,8 @@
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.FsShell;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.FileInputFormat;
@@ -48,6 +50,7 @@
 import edu.uci.ics.hyracks.hdfs.scheduler.Scheduler;
 import edu.uci.ics.hyracks.hdfs.utils.HyracksUtils;
 import edu.uci.ics.hyracks.hdfs.utils.TestUtils;
+import edu.uci.ics.pregelix.core.jobgen.clusterconfig.ClusterConfig;
 
 public class JobRunTestCase {
 	private static final String ACTUAL_RESULT_DIR = "actual";
@@ -57,7 +60,7 @@
 	private static final String HDFS_INPUT_PATH = "/webmap";
 	private static final String HDFS_OUTPUT_PATH = "/webmap_result/";
 
-    private static final String DUMPED_RESULT = ACTUAL_RESULT_DIR + HDFS_OUTPUT_PATH + "/part-00000";
+    private static final String DUMPED_RESULT = ACTUAL_RESULT_DIR + HDFS_OUTPUT_PATH + "/merged.txt";
     private static final String EXPECTED_PATH = "src/test/resources/expected/result2";
 
 	private static final String HYRACKS_APP_NAME = "genomix";
@@ -83,6 +86,7 @@
 		FileInputFormat.setInputPaths(conf, HDFS_INPUT_PATH);
 		FileOutputFormat.setOutputPath(conf, new Path(HDFS_OUTPUT_PATH));
 
+		conf.setInt(GenomixJob.KMER_LENGTH, 5);
 		driver = new Driver(HyracksUtils.CC_HOST,
 				HyracksUtils.TEST_HYRACKS_CC_CLIENT_PORT,
 				numPartitionPerMachine);
@@ -134,11 +138,12 @@
 	public void TestExternalGroupby() throws Exception {
 		cleanUpReEntry();
 		conf.set(GenomixJob.GROUPBY_TYPE, "external");
+		conf.set(GenomixJob.OUTPUT_FORMAT, "text");
 		driver.runJob(new GenomixJob(conf), Plan.BUILD_DEBRUJIN_GRAPH, true);
 		Assert.assertEquals(true, checkResults());
 	}
 
-	@Test
+	//@Test
 	public void TestPreClusterGroupby() throws Exception {
 		cleanUpReEntry();
 		conf.set(GenomixJob.GROUPBY_TYPE, "precluster");
@@ -146,20 +151,17 @@
 		Assert.assertEquals(true, checkResults());
 	}
 
-	@Test
+	//@Test
 	public void TestHybridGroupby() throws Exception {
 		cleanUpReEntry();
 		conf.set(GenomixJob.GROUPBY_TYPE, "hybrid");
+		conf.set(GenomixJob.OUTPUT_FORMAT, "text");
 		driver.runJob(new GenomixJob(conf), Plan.BUILD_DEBRUJIN_GRAPH, true);
 		Assert.assertEquals(true, checkResults());
 	}
 
 	private boolean checkResults() throws Exception {
-		FileSystem dfs = FileSystem.get(conf);
-		Path result = new Path(HDFS_OUTPUT_PATH);
-		Path actual = new Path(ACTUAL_RESULT_DIR);
-		dfs.copyToLocalFile(result, actual);
-
+		FileUtil.copyMerge(FileSystem.get(conf), new Path(HDFS_OUTPUT_PATH), FileSystem.getLocal(new Configuration()), new Path(DUMPED_RESULT), false, conf, null);
 		TestUtils.compareWithResult(new File(EXPECTED_PATH
 				), new File(DUMPED_RESULT));
 		return true;
diff --git a/genomix/genomix-core/src/test/resources/data/0/text.txt b/genomix/genomix-core/src/test/resources/data/0/text.txt
new file mode 100755
index 0000000..f63a141
--- /dev/null
+++ b/genomix/genomix-core/src/test/resources/data/0/text.txt
@@ -0,0 +1,4 @@
+@625E1AAXX100810:1:100:10000:10271/1
+AATAGAAG
++
+EDBDB?BEEEDGGEGGGDGGGA>DG@GGD;GD@DG@F?<B<BFFD?