converter done

Former-commit-id: f512bea019378a3d133faaca57d5f4fef448d4c3

Author: bbengfort

README.md

@@ -1,7 +1,7 @@
 Brisera
 =======
 
-A Python implementation of a distributed seed and reduce algorithm (similar to BlastReduce and CloudBurst) that utilizes RDDs (resilient distributed datasets) to perform fast iterative analyses and dynamic programming without relying on chained MapReduce jobs. 
+A Python implementation of a distributed seed and reduce algorithm (similar to BlastReduce and CloudBurst) that utilizes RDDs (resilient distributed datasets) to perform fast iterative analyses and dynamic programming without relying on chained MapReduce jobs.
 
 Quick Start
 -----------
@@ -20,9 +20,18 @@ To install the required dependencies:
 
 The code for Brisera is found in the `brisera` Python module. This module must be available to the spark applications (e.g. able to be imported) either by running the spark applications locally in the working directory that contains `brisera` or by using a virtual environment (recommended). You can install `brisera` and all dependencies, use the setup.py function:
 
-    $ python setup.py install 
+    $ python setup.py install
 
-But note that you will still have to have access to the Spark applications that are in the `apps/` directory - don't delete them out of hand! 
+But note that you will still have to have access to the Spark applications that are in the `apps/` directory - don't delete them out of hand!
+
+Usage
+-----
+
+To read a burst sequence file (e.g. `fixtures/cloudburst/100k.br`) in order to compare results from CloudBurst to Brisera, you can use the `read_burst.py` Spark application as follows:
+
+    $ spark-submit --master local[*] apps/read_burst.py <sequence_file> <output_dir>
+
+This will write out each record (or chunk) from the sequence file to a text file on disk.
 
 Other Details
 -------------
@@ -37,4 +46,4 @@ Brisera means to "explode" or to "burst" in Swedish. Since I'm reworking CloudBu
 
 1. X\. Li, W. Jiang, Y. Jiang, and Q. Zou, “Hadoop Applications in Bioinformatics,” in Open Cirrus Summit (OCS), 2012 Seventh, 2012, pp. 48–52.
 
-1. R\. K. Menon, G. P. Bhat, and M. C. Schatz, “Rapid parallel genome indexing with MapReduce,” in Proceedings of the second international workshop on MapReduce and its applications, 2011, pp. 51–58.  
+1. R\. K. Menon, G. P. Bhat, and M. C. Schatz, “Rapid parallel genome indexing with MapReduce,” in Proceedings of the second international workshop on MapReduce and its applications, 2011, pp. 51–58.

apps/convert_fasta.py

@@ -7,9 +7,12 @@
 ## Imports
 ##########################################################################
 
+import os
 import sys
+import shutil
+import tempfile
 
-from operator import add
+from brisera.convert import FastaChunker
 from pyspark import SparkConf, SparkContext
 
 if __name__ == "__main__":
@@ -23,7 +26,25 @@
 
     infile  = sys.argv[1]
     outfile = sys.argv[2]
+    tempdir = tempfile.mkdtemp(prefix="fasta")
+    tempout = os.path.join(tempdir, "output")
 
     print "Converting FASTA %s to Sequence %s" % (infile, outfile)
 
+    chunker = FastaChunker(infile)
+    chunks  = sc.parallelize(chunker.convert()).coalesce(1, shuffle=True)
+    chunks.saveAsSequenceFile(tempout)
 
+    partfile = None
+    for name in os.listdir(tempout):
+        if name.startswith('part-'):
+            partfile = os.path.join(tempout, name)
+            break
+
+    if partfile is None:
+        raise Exception("Could not find partition file in %s!" % tempout)
+
+    shutil.move(partfile, outfile)
+    shutil.rmtree(tempdir)
+
+    assert not os.path.exists(tempdir)

brisera/convert.py

@@ -2,6 +2,8 @@
 Handles the conversion of a FASTA sequence into a sequence format
 """
 
+import cPickle
+
 from brisera.utils import fasta
 from brisera.config import settings
 
@@ -50,18 +52,15 @@ def chunk(self, sequence):
             else:
                 offset = end - settings.overlap
 
-    def convert(self, writer):
+    def convert(self):
         """
-        The main entry point, convert the FASTA file and output it by
-        writing it to the given stream (the writer).
+        The main entry point, convert the FASTA file and yielding pairs
+        where the key is the index and the value is the record.
         """
 
         for idx, seq in self:
-            for chunk in self.chunk(seq):
-                record = str((idx, chunk))
-                writer.write(record+"\n")
-                break
-            break
+            for record in self.chunk(seq):
+                yield (idx, cPickle.dumps(record, cPickle.HIGHEST_PROTOCOL))
 
     def __iter__(self):
         """