Map/Reduce version in pregel is completed
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.2.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.2.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.2.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.3.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.3.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.3.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.4.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.4.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.4.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.5.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.5.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.5.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.6.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.6.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.6.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.7.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.7.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.7.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.8.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.8.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.8.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.9.crc b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.9.crc
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/.9.crc
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/2 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/2
new file mode 100755
index 0000000..f8bec75
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/2
@@ -0,0 +1,2 @@
+ATA {[(1-2_1)] [] [] [] [] AATA}
+AAT {[(1-1_1)] [] [] [] [] AATA}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/3 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/3
new file mode 100755
index 0000000..0f3d880
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/3
@@ -0,0 +1 @@
+ATA {[(1-2_1)] [] [] [] [] AATAG}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/4 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/4
new file mode 100755
index 0000000..91211fc
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/4
@@ -0,0 +1,2 @@
+CTA {[(1-3_1)] [] [] [] [] TCTATT}
+ATA {[(1-2_1)] [] [] [] [] AATAGA}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/5 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/5
new file mode 100755
index 0000000..a2ae06f
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/5
@@ -0,0 +1,2 @@
+ATA {[(1-2_1)] [] [] [] [] AATAGAA}
+AGA {[(1-4_1)] [] [] [] [] AATAGAA}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/6 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/6
new file mode 100755
index 0000000..c084bee
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/6
@@ -0,0 +1,2 @@
+CTA {[(1-3_1)] [] [] [] [] GTTCTATT}
+AGA {[(1-4_1)] [] [] [] [] AATAGAAC}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/7 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/7
new file mode 100755
index 0000000..30b31af
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/7
@@ -0,0 +1 @@
+AGA {[(1-4_1)] [] [] [] [] AATAGAACT}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/8 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/8
new file mode 100755
index 0000000..900707b
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/8
@@ -0,0 +1,2 @@
+GAA {[(1-5_1)] [] [] [] [] AATAGAACTT}
+AGA {[(1-4_1)] [] [] [] [] AATAGAACTT}
diff --git a/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/9 b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/9
new file mode 100755
index 0000000..862bc7c4
--- /dev/null
+++ b/genomix/genomix-pregelix/data/LogAlgorithmForMergeGraph/txt/9
@@ -0,0 +1,2 @@
+AGA {[(1-4_1)] [] [] [] [] AATAGAACTTA}
+AAC {[(1-6_1)] [] [] [] [] AATAGAACTTA}
diff --git a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/io/VertexValueWritable.java b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/io/VertexValueWritable.java
index 48a9f52..f1ed1ad 100644
--- a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/io/VertexValueWritable.java
+++ b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/io/VertexValueWritable.java
@@ -11,7 +11,18 @@
public class VertexValueWritable implements WritableComparable<VertexValueWritable> {
- public static class VertexStateFlag {
+ public static class State extends VertexStateFlag{
+ public static final byte HEAD_SHOULD_MERGEWITHPREV = 0b101 << 0;
+ public static final byte HEAD_SHOULD_MERGEWITHNEXT = 0b111 << 0;
+
+ public static final byte NO_MERGE = 0b00 << 3;
+ public static final byte SHOULD_MERGEWITHNEXT = 0b01 << 3;
+ public static final byte SHOULD_MERGEWITHPREV = 0b10 << 3;
+ public static final byte SHOULD_MERGE_MASK = 0b11 << 3;
+ public static final byte SHOULD_MERGE_CLEAR = 0b1100111;
+ }
+
+ public static class VertexStateFlag extends FakeFlag {
public static final byte IS_NON = 0b00 << 5;
public static final byte IS_RANDOMTAIL = 0b00 << 5;
public static final byte IS_HEAD = 0b01 << 5;
@@ -23,15 +34,9 @@
public static final byte VERTEX_CLEAR = (byte) 11001111;
}
- public static class State extends VertexStateFlag{
- public static final byte HEAD_SHOULD_MERGEWITHPREV = 0b101 << 0;
- public static final byte HEAD_SHOULD_MERGEWITHNEXT = 0b111 << 0;
-
- public static final byte NO_MERGE = 0b00 << 3;
- public static final byte SHOULD_MERGEWITHNEXT = 0b01 << 3;
- public static final byte SHOULD_MERGEWITHPREV = 0b10 << 3;
- public static final byte SHOULD_MERGE_MASK = 0b11 << 3;
- public static final byte SHOULD_MERGE_CLEAR = 0b1100111;
+ public static class FakeFlag{
+ public static final byte IS_NONFAKE = 0 << 0;
+ public static final byte IS_FAKE = 1 << 0;
}
private PositionListWritable nodeIdList;
diff --git a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/BasicPathMergeVertex.java b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/BasicPathMergeVertex.java
index 821117b..3039bc1 100644
--- a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/BasicPathMergeVertex.java
+++ b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/BasicPathMergeVertex.java
@@ -415,8 +415,6 @@
sendMsg(incomingMsg.getSourceVertexId(), outgoingMsg); //getPreDestVertexId(getVertexValue())
break;
}
-// if(headBecomeOldHead)
-// getVertexValue().processDelete(neighborToMeDir, incomingMsg.getSourceVertexId());
}
/**
diff --git a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/MapReduceVertex.java b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/MapReduceVertex.java
index 0cc198b..5f70772 100644
--- a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/MapReduceVertex.java
+++ b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/MapReduceVertex.java
@@ -3,12 +3,14 @@
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
+import java.util.Random;
import edu.uci.ics.genomix.pregelix.client.Client;
import edu.uci.ics.genomix.pregelix.format.GraphCleanInputFormat;
-import edu.uci.ics.genomix.pregelix.format.GraphCleanOutputFormat;
+import edu.uci.ics.genomix.pregelix.format.P2PathMergeOutputFormat;
import edu.uci.ics.genomix.pregelix.io.MessageWritable;
import edu.uci.ics.genomix.pregelix.io.VertexValueWritable;
+import edu.uci.ics.genomix.pregelix.io.VertexValueWritable.State;
import edu.uci.ics.genomix.pregelix.type.MessageFlag;
import edu.uci.ics.genomix.type.KmerBytesWritable;
import edu.uci.ics.genomix.type.KmerListWritable;
@@ -19,45 +21,102 @@
public class MapReduceVertex extends
BasicPathMergeVertex {
- KmerListWritable kmerList = new KmerListWritable(kmerSize);
- Map<KmerBytesWritable, KmerListWritable> kmerMapper = new HashMap<KmerBytesWritable, KmerListWritable>();
+ private static boolean fakeVertexExist = false;
+ private static KmerBytesWritable fakeVertex = null;
+
+ private KmerBytesWritable reverseKmer;
+ private KmerListWritable kmerList = null;
+ private Map<KmerBytesWritable, KmerListWritable> kmerMapper = new HashMap<KmerBytesWritable, KmerListWritable>();
+
+ /**
+ * initiate kmerSize, maxIteration
+ */
+ public void initVertex() {
+ if (kmerSize == -1)
+ kmerSize = getContext().getConfiguration().getInt(KMER_SIZE, 5);
+ if (maxIteration < 0)
+ maxIteration = getContext().getConfiguration().getInt(ITERATIONS, 1000000);
+ if(incomingMsg == null)
+ incomingMsg = new MessageWritable(kmerSize);
+ if(outgoingMsg == null)
+ outgoingMsg = new MessageWritable(kmerSize);
+ else
+ outgoingMsg.reset(kmerSize);
+ if(reverseKmer == null)
+ reverseKmer = new KmerBytesWritable(kmerSize);
+ if(kmerList == null)
+ kmerList = new KmerListWritable(kmerSize);
+ else
+ kmerList.reset(kmerSize);
+ if(fakeVertex == null){
+ fakeVertex = new KmerBytesWritable(kmerSize + 1);
+ String random = generaterRandomString(kmerSize + 1);
+ fakeVertex.setByRead(random.getBytes(), 0);
+ }
+ }
+
+ /**
+ * Generate random string from [ACGT]
+ */
+ public String generaterRandomString(int n){
+ char[] chars = "ACGT".toCharArray();
+ StringBuilder sb = new StringBuilder();
+ Random random = new Random();
+ for (int i = 0; i < n; i++) {
+ char c = chars[random.nextInt(chars.length)];
+ sb.append(c);
+ }
+ return sb.toString();
+ }
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public void compute(Iterator<MessageWritable> msgIterator) {
+ initVertex();
if(getSuperstep() == 1){
- //add a fake vertex
- Vertex vertex = (Vertex) BspUtils.createVertex(getContext().getConfiguration());
- vertex.getMsgList().clear();
- vertex.getEdges().clear();
- KmerBytesWritable vertexId = new KmerBytesWritable();
- VertexValueWritable vertexValue = new VertexValueWritable();
- vertexId.setByRead(("XXX").getBytes(), 0);
- vertexValue.setFakeVertex(true);
- vertex.setVertexValue(vertexValue);
-
- addVertex(vertexId, vertex);
+ if(!fakeVertexExist){
+ //add a fake vertex
+ Vertex vertex = (Vertex) BspUtils.createVertex(getContext().getConfiguration());
+ vertex.getMsgList().clear();
+ vertex.getEdges().clear();
+ VertexValueWritable vertexValue = new VertexValueWritable(kmerSize + 1);
+ vertexValue.setState(State.IS_FAKE);
+ vertexValue.setFakeVertex(true);
+
+ vertex.setVertexId(fakeVertex);
+ vertex.setVertexValue(vertexValue);
+
+ addVertex(fakeVertex, vertex);
+ fakeVertexExist = true;
+ }
}
else if(getSuperstep() == 2){
if(!getVertexValue().isFakeVertex()){
- destVertexId.setByRead(("XXX").getBytes(), 0);
outgoingMsg.setSourceVertexId(getVertexId());
outgoingMsg.setAcutalKmer(getVertexValue().getKmer());
- sendMsg(destVertexId, outgoingMsg);
+ sendMsg(fakeVertex, outgoingMsg);
+ voteToHalt();
}
- voteToHalt();
} else if(getSuperstep() == 3){
kmerMapper.clear();
/** Mapper **/
while(msgIterator.hasNext()){
incomingMsg = msgIterator.next();
- tmpKmer.set(incomingMsg.getActualKmer());
+ String kmerString = incomingMsg.getActualKmer().toString();
+ tmpKmer.reset(kmerString.length());
+ reverseKmer.reset(kmerString.length());
+ tmpKmer.setByRead(kmerString.getBytes(), 0);
+ reverseKmer.setByReadReverse(kmerString.getBytes(), 0);
+
+ if(reverseKmer.compareTo(tmpKmer) < 0)
+ tmpKmer.set(reverseKmer);
if(!kmerMapper.containsKey(tmpKmer)){
kmerList.reset();
kmerList.append(incomingMsg.getSourceVertexId());
kmerMapper.put(tmpKmer, kmerList);
} else{
kmerList.set(kmerMapper.get(tmpKmer));
+ kmerList.append(incomingMsg.getSourceVertexId());
kmerMapper.put(tmpKmer, kmerList);
}
}
@@ -67,9 +126,16 @@
for(int i = 1; i < kmerList.getCountOfPosition(); i++){
//send kill message
outgoingMsg.setFlag(MessageFlag.KILL);
+ destVertexId.set(kmerList.getPosition(i));
+ sendMsg(destVertexId, outgoingMsg);
}
}
} else if(getSuperstep() == 4){
+ /** only for test single MapReduce job**/
+ if(!msgIterator.hasNext() && getVertexValue().getState() == State.IS_FAKE){
+ fakeVertexExist = false;
+ deleteVertex(fakeVertex);
+ }
while(msgIterator.hasNext()){
incomingMsg = msgIterator.next();
inFlag = incomingMsg.getFlag();
@@ -89,7 +155,7 @@
* BinaryInput and BinaryOutput
*/
job.setVertexInputFormatClass(GraphCleanInputFormat.class);
- job.setVertexOutputFormatClass(GraphCleanOutputFormat.class);
+ job.setVertexOutputFormatClass(P2PathMergeOutputFormat.class);
job.setDynamicVertexValueSize(true);
job.setOutputKeyClass(KmerBytesWritable.class);
job.setOutputValueClass(VertexValueWritable.class);
diff --git a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/P4ForPathMergeVertex.java b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/P4ForPathMergeVertex.java
index e3af9bf..a14be92 100644
--- a/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/P4ForPathMergeVertex.java
+++ b/genomix/genomix-pregelix/src/main/java/edu/uci/ics/genomix/pregelix/operator/pathmerge/P4ForPathMergeVertex.java
@@ -93,7 +93,6 @@
inFlag = (byte)0;
// Node may be marked as head b/c it's a real head or a real tail
headFlag = (byte) (State.IS_HEAD & getVertexValue().getState());
- outgoingMsg.reset();
}
protected boolean isNodeRandomHead(KmerBytesWritable nodeKmer) {
diff --git a/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobGen/JobGenerator.java b/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobGen/JobGenerator.java
index eb17734..6e43552 100644
--- a/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobGen/JobGenerator.java
+++ b/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobGen/JobGenerator.java
@@ -100,7 +100,7 @@
PregelixJob job = new PregelixJob(jobName);
job.setVertexClass(MapReduceVertex.class);
job.setVertexInputFormatClass(GraphCleanInputFormat.class);
- job.setVertexOutputFormatClass(GraphCleanOutputFormat.class);
+ job.setVertexOutputFormatClass(P2PathMergeOutputFormat.class);
job.setDynamicVertexValueSize(true);
job.setOutputKeyClass(KmerBytesWritable.class);
job.setOutputValueClass(VertexValueWritable.class);
@@ -109,7 +109,7 @@
}
private static void genMapReduceGraph() throws IOException {
- generateMapReduceGraphJob("LogAlgorithmForMergeGraph", outputBase + "LogAlgorithmForMergeGraph.xml");
+ generateMapReduceGraphJob("MapReduceGraph", outputBase + "MapReduceGraph.xml");
}
// private static void generateTipAddGraphJob(String jobName, String outputPath) throws IOException {
// PregelixJob job = new PregelixJob(jobName);
diff --git a/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobRun/PathMergeSmallTestSuite.java b/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobRun/PathMergeSmallTestSuite.java
index 115f090..8172fb6 100644
--- a/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobRun/PathMergeSmallTestSuite.java
+++ b/genomix/genomix-pregelix/src/test/java/edu/uci/ics/genomix/pregelix/JobRun/PathMergeSmallTestSuite.java
@@ -43,7 +43,7 @@
public class PathMergeSmallTestSuite extends TestSuite {
private static final Logger LOGGER = Logger.getLogger(PathMergeSmallTestSuite.class.getName());
- public static final String PreFix = "data/PathMergeTestSet"; //"graphbuildresult";
+ public static final String PreFix = "data/LogAlgorithmForMergeGraph/bin"; //"graphbuildresult";
public static final String[] TestDir = { PreFix + File.separator
+ "2", PreFix + File.separator
+ "3", PreFix + File.separator
diff --git a/genomix/genomix-pregelix/src/test/resources/only_pathmerge.txt b/genomix/genomix-pregelix/src/test/resources/only_pathmerge.txt
index 5a15ca0..710a096 100644
--- a/genomix/genomix-pregelix/src/test/resources/only_pathmerge.txt
+++ b/genomix/genomix-pregelix/src/test/resources/only_pathmerge.txt
@@ -1 +1 @@
-LogAlgorithmForMergeGraph.xml
+MapReduceGraph.xml