10mers.adam

time{
import org.bdgenomics.adam.rdd.ADAMContext
import org.bdgenomics.adam.projections.{AlignmentRecordField, Projection}

val ac = new ADAMContext(sc)
val reads = ac.loadAlignments(
  "/home/deploy/data/HG00096.unmapped.ILLUMINA.bwa.GBR.low_coverage.20120522.adam",
  projection = Some(
    Projection(
      AlignmentRecordField.sequence,
      AlignmentRecordField.readMapped,
      AlignmentRecordField.mapq
    )
  )
)

val kmers =
  reads
    .rdd
    .flatMap(_.getSequence.sliding(10).map(k => (k, 1L)))
    .reduceByKey(_ + _)
    .map(_.swap)
    .sortByKey(ascending = false)

kmers.take(10).foreach(println)
}

def time[T](block: => T): T = {
val start = System.currentTimeMillis
val res = block
val totalTime = System.currentTimeMillis - start
println("Elapsed time: %1d ms".format(totalTime))
res
}

time{
import org.bdgenomics.adam.rdd.ADAMContext
import org.bdgenomics.adam.projections.{AlignmentRecordField, Projection}

val ac = new ADAMContext(sc)
val reads = ac.loadAlignments(
"/home/deploy/data/HG00096.unmapped.ILLUMINA.bwa.GBR.low_coverage.20120522.adam",
projection = Some(
Projection(
AlignmentRecordField.sequence,
AlignmentRecordField.readMapped,
AlignmentRecordField.mapq
)
)
)

val script = "/home/deploy/mapv2/mapv2"
val result = reads.rdd .map(x => x.getSequence).pipe(script).map(.split(',') match {case Array(x,y) => (x, y.toLong)}).reduceByKey( + _)

result.take(10).foreach(println)
}