Skip to content

kill Job arg to ADAMSparkCommand #670

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

kill Job arg to ADAMSparkCommand #670

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/ADAM2Vcf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ class ADAM2VcfArgs extends Args4jBase with ParquetArgs {
class ADAM2Vcf(val args: ADAM2VcfArgs) extends ADAMSparkCommand[ADAM2VcfArgs] with DictionaryCommand with Logging {
val companion = ADAM2Vcf

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
var dictionary: Option[SequenceDictionary] = loadSequenceDictionary(args.dictionaryFile)
if (dictionary.isDefined)
log.info("Using contig translation")
Expand Down
7 changes: 2 additions & 5 deletions adam-cli/src/main/scala/org/bdgenomics/adam/cli/ADAMCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,8 @@ package org.bdgenomics.adam.cli

import java.io.{ StringWriter, PrintWriter }

import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{ SparkConf, Logging, SparkContext }
import org.bdgenomics.utils.instrumentation._
import org.bdgenomics.adam.util.HadoopUtil

trait ADAMCommandCompanion {
val commandName: String
Expand All @@ -43,7 +41,7 @@ trait ADAMCommand extends Runnable {
trait ADAMSparkCommand[A <: Args4jBase] extends ADAMCommand with Logging {
protected val args: A

def run(sc: SparkContext, job: Job)
def run(sc: SparkContext): Unit

def run() {
val start = System.nanoTime()
Expand All @@ -52,9 +50,8 @@ trait ADAMSparkCommand[A <: Args4jBase] extends ADAMCommand with Logging {
conf.setMaster("local[%d]".format(Runtime.getRuntime.availableProcessors()))
}
val sc = new SparkContext(conf)
val job = HadoopUtil.newJob()
val metricsListener = initializeMetrics(sc)
run(sc, job)
run(sc)
val totalTime = System.nanoTime() - start
printMetrics(totalTime, metricsListener)
}
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/Adam2Fastq.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ object Adam2Fastq extends ADAMCommandCompanion {
class Adam2Fastq(val args: Adam2FastqArgs) extends ADAMSparkCommand[Adam2FastqArgs] {
override val companion = Adam2Fastq

override def run(sc: SparkContext, job: Job): Unit = {
override def run(sc: SparkContext): Unit = {

val projectionOpt =
if (!args.disableProjection)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/AlleleCount.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ object AlleleCountHelper extends Serializable {
class AlleleCount(val args: AlleleCountArgs) extends ADAMSparkCommand[AlleleCountArgs] with Logging {
val companion = AlleleCount

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {

val adamVariants: RDD[Genotype] = sc.loadGenotypes(args.adamFile)
AlleleCountHelper.countAlleles(adamVariants, args)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/CalculateDepth.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ class CalculateDepthArgs extends Args4jBase with ParquetArgs {
class CalculateDepth(protected val args: CalculateDepthArgs) extends ADAMSparkCommand[CalculateDepthArgs] {
val companion: ADAMCommandCompanion = CalculateDepth

def run(sc: SparkContext, job: Job): Unit = {
def run(sc: SparkContext): Unit = {

val proj = Projection(contig, start, cigar, readMapped)

Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/CountContigKmers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ class CountContigKmersArgs extends Args4jBase with ParquetArgs {
class CountContigKmers(protected val args: CountContigKmersArgs) extends ADAMSparkCommand[CountContigKmersArgs] with Logging {
val companion = CountContigKmers

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {

// Quiet Parquet...
ParquetLogger.hadoopLoggerLevel(Level.SEVERE)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/CountReadKmers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ class CountReadKmersArgs extends Args4jBase with ParquetArgs {
class CountReadKmers(protected val args: CountReadKmersArgs) extends ADAMSparkCommand[CountReadKmersArgs] with Logging {
val companion = CountReadKmers

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {

// Quiet Parquet...
ParquetLogger.hadoopLoggerLevel(Level.SEVERE)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/Fasta2ADAM.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ class Fasta2ADAMArgs extends Args4jBase with ParquetSaveArgs {
class Fasta2ADAM(protected val args: Fasta2ADAMArgs) extends ADAMSparkCommand[Fasta2ADAMArgs] with Logging {
val companion = Fasta2ADAM

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
log.info("Loading FASTA data from disk.")
val adamFasta = sc.loadFasta(args.fastaFile, fragmentLength = args.fragmentLength)

Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/Features2ADAM.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ class Features2ADAM(val args: Features2ADAMArgs)
extends ADAMSparkCommand[Features2ADAMArgs] {
val companion = Features2ADAM

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
sc.loadFeatures(args.featuresFile).adamParquetSave(args)
}
}
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/FlagStat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ class FlagStatArgs extends Args4jBase with ParquetArgs {
class FlagStat(protected val args: FlagStatArgs) extends ADAMSparkCommand[FlagStatArgs] {
val companion: ADAMCommandCompanion = FlagStat

def run(sc: SparkContext, job: Job): Unit = {
def run(sc: SparkContext): Unit = {

val projection = Projection(
AlignmentRecordField.readMapped,
Expand Down
5 changes: 2 additions & 3 deletions adam-cli/src/main/scala/org/bdgenomics/adam/cli/Flatten.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package org.bdgenomics.adam.cli

import org.apache.avro.Schema
import org.apache.avro.generic.IndexedRecord
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.rdd.MetricsContext._
import org.apache.spark.{ Logging, SparkContext }
import org.bdgenomics.adam.rdd.ADAMContext._
Expand Down Expand Up @@ -51,7 +50,7 @@ class FlattenArgs extends Args4jBase with ParquetSaveArgs {
class Flatten(val args: FlattenArgs) extends ADAMSparkCommand[FlattenArgs] with Logging {
val companion = Flatten

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {

val job = HadoopUtil.newJob(sc)
val records = sc.newAPIHadoopFile(
Expand All @@ -76,4 +75,4 @@ class Flatten(val args: FlattenArgs) extends ADAMSparkCommand[FlattenArgs] with
args.disableDictionaryEncoding,
Some(flatSchema))
}
}
}
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/ListDict.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ class ListDictArgs extends Args4jBase with ParquetArgs {
class ListDict(protected val args: ListDictArgs) extends ADAMSparkCommand[ListDictArgs] {
val companion: ADAMCommandCompanion = ListDict

def run(sc: SparkContext, job: Job): Unit = {
def run(sc: SparkContext): Unit = {
val dict = sc.adamDictionaryLoad[AlignmentRecord](args.inputPath)

dict.records.foreach {
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/PluginExecutor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ class PluginExecutor(protected val args: PluginExecutorArgs) extends ADAMSparkCo
output.map(_.toString).collect().foreach(println)
}

def run(sc: SparkContext, job: Job): Unit = {
def run(sc: SparkContext): Unit = {
val plugin = loadPlugin[AlignmentRecord, Any](args.plugin)
val accessControl = loadAccessControl[AlignmentRecord](args.accessControl)

Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/PrintADAM.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ class PrintADAM(protected val args: PrintADAMArgs) extends ADAMSparkCommand[Prin
})
}

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
val output = Option(args.outputFile)
args.filesToPrint.foreach(file => {
displayRaw(sc, file, pretty = args.prettyRaw, output = output)
Expand Down
4 changes: 2 additions & 2 deletions adam-cli/src/main/scala/org/bdgenomics/adam/cli/PrintGenes.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ object PrintGenes extends ADAMCommandCompanion {
val commandName: String = "print_genes"
val commandDescription: String = "Load a GTF file containing gene annotations and print the corresponding gene models"

def apply(cmdLine: Array[String]) = {
def apply(cmdLine: Array[String]): ADAMCommand = {
new PrintGenes(Args4j[PrintGenesArgs](cmdLine))
}
}
Expand All @@ -44,7 +44,7 @@ class PrintGenes(protected val args: PrintGenesArgs)

val companion = PrintGenes

def run(sc: SparkContext, job: Job): Unit = {
def run(sc: SparkContext): Unit = {
val genes: RDD[Gene] = sc.loadGenes(args.gtfInput)

genes.map(printGene).collect().foreach(println)
Expand Down
10 changes: 5 additions & 5 deletions adam-cli/src/main/scala/org/bdgenomics/adam/cli/PrintTags.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import org.bdgenomics.adam.projections.AlignmentRecordField._
import org.bdgenomics.adam.projections.Projection
import org.bdgenomics.adam.rdd.ADAMContext._
import org.bdgenomics.formats.avro.AlignmentRecord
import org.kohsuke.args4j.{ Argument, Option }
import org.kohsuke.args4j.{ Argument, Option => A4JOption }

/**
* Reads in the tagStrings field of every record, and prints out the set of unique
Expand All @@ -44,20 +44,20 @@ class PrintTagsArgs extends Args4jBase with ParquetArgs {
@Argument(required = true, metaVar = "INPUT", usage = "The ADAM file to scan for tags", index = 0)
val inputPath: String = null

@Option(required = false, name = "-list",
@A4JOption(required = false, name = "-list",
usage = "When value is set to <N>, also lists the first N attribute fields for ADAMRecords in the input")
var list: String = null

@Option(required = false, name = "-count",
@A4JOption(required = false, name = "-count",
usage = "comma-separated list of tag names; for each tag listed, we print the distinct values and their counts")
var count: String = null

}

class PrintTags(protected val args: PrintTagsArgs) extends ADAMSparkCommand[PrintTagsArgs] {
class PrintTags(protected val args: PrintTagsArgs) extends ADAMSparkCommand[PrintTagsArgs] with Serializable {
val companion: ADAMCommandCompanion = PrintTags

def run(sc: SparkContext, job: Job): Unit = {
override def run(sc: SparkContext): Unit = {
val toCount = if (args.count != null) args.count.split(",").toSet else Set()

val proj = Projection(attributes, primaryAlignment, readMapped, readPaired, failedVendorQualityChecks)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/Transform.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@ class Transform(protected val args: TransformArgs) extends ADAMSparkCommand[Tran
adamRecords
}

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
this.apply({
if (args.forceLoadBam) {
sc.loadBam(args.inputPath)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/Vcf2ADAM.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ class Vcf2ADAMArgs extends Args4jBase with ParquetSaveArgs {
class Vcf2ADAM(val args: Vcf2ADAMArgs) extends ADAMSparkCommand[Vcf2ADAMArgs] with DictionaryCommand with Logging {
val companion = Vcf2ADAM

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {

var dictionary: Option[SequenceDictionary] = loadSequenceDictionary(args.dictionaryFile)
if (dictionary.isDefined)
Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/VcfAnnotation2ADAM.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ class VcfAnnotation2ADAMArgs extends Args4jBase with ParquetSaveArgs {
class VcfAnnotation2ADAM(val args: VcfAnnotation2ADAMArgs) extends ADAMSparkCommand[VcfAnnotation2ADAMArgs] with Logging {
val companion = VcfAnnotation2ADAM

def run(sc: SparkContext, job: Job) {
def run(sc: SparkContext) {
log.info("Reading VCF file from %s".format(args.vcfFile))
val annotations: RDD[DatabaseVariantAnnotation] = sc.loadVcfAnnotations(args.vcfFile)

Expand Down
2 changes: 1 addition & 1 deletion adam-cli/src/main/scala/org/bdgenomics/adam/cli/View.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ class View(val args: ViewArgs) extends ADAMSparkCommand[ViewArgs] {
reads
}

def run(sc: SparkContext, job: Job) = {
def run(sc: SparkContext) = {

val reads: RDD[AlignmentRecord] = applyFilters(sc.loadAlignments(args.inputPath))

Expand Down
8 changes: 3 additions & 5 deletions adam-cli/src/test/scala/org/bdgenomics/adam/cli/Features2ADAMSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import java.io._

import org.bdgenomics.adam.projections.Projection
import org.bdgenomics.adam.projections.FeatureField._
import org.bdgenomics.adam.util.{ HadoopUtil, ADAMFunSuite }
import org.bdgenomics.adam.util.ADAMFunSuite
import org.bdgenomics.formats.avro.Feature

class Features2ADAMSuite extends ADAMFunSuite {
Expand All @@ -42,8 +42,7 @@ class Features2ADAMSuite extends ADAMFunSuite {
val args: Features2ADAMArgs = Args4j.apply[Features2ADAMArgs](argLine)

val features2Adam = new Features2ADAM(args)
val job = HadoopUtil.newJob()
features2Adam.run(sc, job)
features2Adam.run(sc)

val schema = Projection(featureId, contig, start, strand)
val lister = new ParquetLister[Feature](Some(schema))
Expand Down Expand Up @@ -88,8 +87,7 @@ class Features2ADAMSuite extends ADAMFunSuite {
val adamArgLine = "%s %s".format(bedPath, outputPath).split("\\s+")
val adamArgs: Features2ADAMArgs = Args4j.apply[Features2ADAMArgs](adamArgLine)
val features2Adam = new Features2ADAM(adamArgs)
val job = HadoopUtil.newJob()
features2Adam.run(sc, job)
features2Adam.run(sc)

val schema = Projection(featureId, contig, start, end, value)
val lister = new ParquetLister[Feature](Some(schema))
Expand Down
8 changes: 3 additions & 5 deletions adam-cli/src/test/scala/org/bdgenomics/adam/cli/FlattenSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ package org.bdgenomics.adam.cli
import java.io._

import org.apache.avro.generic.GenericRecord
import org.bdgenomics.adam.util.{ ADAMFunSuite, HadoopUtil }
import org.bdgenomics.adam.util.ADAMFunSuite
import org.bdgenomics.formats.avro.Genotype

class FlattenSuite extends ADAMFunSuite {
Expand All @@ -40,8 +40,7 @@ class FlattenSuite extends ADAMFunSuite {
val argLine = "%s %s".format(inputPath, outputPath).split("\\s+")
val args: Vcf2ADAMArgs = Args4j.apply[Vcf2ADAMArgs](argLine)
val vcf2Adam = new Vcf2ADAM(args)
val job = HadoopUtil.newJob()
vcf2Adam.run(sc, job)
vcf2Adam.run(sc)

val lister = new ParquetLister[Genotype]()
val records = lister.materialize(outputPath).toSeq
Expand All @@ -54,8 +53,7 @@ class FlattenSuite extends ADAMFunSuite {
val flattenArgLine = "%s %s".format(outputPath, flatPath).split("\\s+")
val flattenArgs: FlattenArgs = Args4j.apply[FlattenArgs](flattenArgLine)
val flatten = new Flatten(flattenArgs)
val flattenJob = HadoopUtil.newJob()
flatten.run(sc, flattenJob)
flatten.run(sc)

val flatLister = new ParquetLister[GenericRecord]()
val flatRecords = flatLister.materialize(flatPath).toSeq
Expand Down
8 changes: 4 additions & 4 deletions adam-cli/src/test/scala/org/bdgenomics/adam/cli/PluginExecutorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.bdgenomics.adam.cli

import java.io._
import org.bdgenomics.adam.util.{ HadoopUtil, ADAMFunSuite }
import org.bdgenomics.adam.util.ADAMFunSuite

class PluginExecutorSuite extends ADAMFunSuite {

Expand All @@ -37,7 +37,7 @@ class PluginExecutorSuite extends ADAMFunSuite {
val pluginExecutor = new PluginExecutor(args)

val bytesWritten = new ByteArrayOutputStream()
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc, HadoopUtil.newJob()))
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc))

val outputString = bytesWritten.toString

Expand All @@ -60,7 +60,7 @@ class PluginExecutorSuite extends ADAMFunSuite {
val pluginExecutor = new PluginExecutor(args)

val bytesWritten = new ByteArrayOutputStream()
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc, HadoopUtil.newJob()))
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc))

val outputString = bytesWritten.toString

Expand All @@ -85,7 +85,7 @@ class PluginExecutorSuite extends ADAMFunSuite {
val pluginExecutor = new PluginExecutor(args)

val bytesWritten = new ByteArrayOutputStream()
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc, HadoopUtil.newJob()))
scala.Console.withOut(bytesWritten)(pluginExecutor.run(sc))

val outputString = bytesWritten.toString

Expand Down
4 changes: 0 additions & 4 deletions .../src/main/scala/org/bdgenomics/adam/rdd/contig/NucleotideContigFragmentRDDFunctions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,6 @@ import org.bdgenomics.adam.converters.FragmentConverter
import org.bdgenomics.adam.models._
import org.bdgenomics.adam.rdd.ADAMContext._
import org.bdgenomics.adam.rdd.ADAMSequenceDictionaryRDDAggregator
import org.bdgenomics.adam.util.{
HadoopUtil,
ParquetLogger
}
import org.bdgenomics.formats.avro._
import parquet.avro.AvroParquetOutputFormat
import parquet.hadoop.ParquetOutputFormat
Expand Down