Merge pull request #146 from tomerk/dag-construction-v2

Dag construction v2

Author: shivaram

src/main/resources/log4j.properties

@@ -6,5 +6,6 @@ log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}:
 
 # Only pay attention to INFO messages from Keystone.
 log4j.logger.pipelines=INFO
+log4j.logger.workflow=INFO
 log4j.logger.nodes=INFO
 log4j.logger.utils=INFO

src/main/scala/loaders/VOCLoader.scala

@@ -1,21 +1,9 @@
 package loaders
 
-import java.awt.image.BufferedImage
-import java.io.{ByteArrayInputStream, InputStream}
-import java.net.URI
-import java.util.zip.GZIPInputStream
-import javax.imageio.ImageIO
-
-import org.apache.commons.compress.archivers.ArchiveStreamFactory
-import org.apache.commons.compress.archivers.tar.TarArchiveInputStream
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
-import pipelines.{Logging, Transformer}
-import utils.{MultiLabeledImage, Image, ImageConversions}
-
-import scala.collection.mutable.ArrayBuffer
+import pipelines.Logging
+import utils.MultiLabeledImage
 
 
 case class VOCDataPath(imagesDirName: String, namePrefix: String, numParts: Option[Int])

src/main/scala/nodes/images/Convolver.scala

@@ -6,6 +6,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import pipelines._
 import utils.{ChannelMajorArrayVectorizedImage, ImageMetadata, _}
+import workflow.Transformer
 
 /**
  * Convolves images with a bank of convolution filters. Convolution filters must be square.

src/main/scala/nodes/images/DaisyExtractor.scala

@@ -3,7 +3,7 @@ package nodes.images
 import breeze.linalg._
 import breeze.numerics._
 
-import pipelines.Transformer
+import workflow.Transformer
 import utils.Image
 import utils.ImageUtils
 

src/main/scala/nodes/images/FisherVector.scala

@@ -1,7 +1,7 @@
 package nodes.images
 
 import breeze.linalg.DenseMatrix
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * Abstract interface for Fisher Vector.

src/main/scala/nodes/images/GrayScaler.scala

@@ -1,6 +1,6 @@
 package nodes.images
 
-import pipelines.Transformer
+import workflow.Transformer
 import utils.{ImageUtils, Image}
 
 /**

src/main/scala/nodes/images/HogExtractor.scala

@@ -18,7 +18,7 @@ package nodes.images
 
 import breeze.linalg._
 
-import pipelines.Transformer
+import workflow.Transformer
 import utils.ChannelMajorArrayVectorizedImage
 import utils.Image
 import utils.ImageUtils

src/main/scala/nodes/images/ImageVectorizer.scala

@@ -4,6 +4,7 @@ import breeze.linalg.DenseVector
 import org.apache.spark.rdd.RDD
 import pipelines._
 import utils.Image
+import workflow.Transformer
 
 /**
  * Takes an image and converts it to a dense vector.

src/main/scala/nodes/images/LCSExtractor.scala

@@ -2,7 +2,7 @@ package nodes.images
 
 import breeze.linalg._
 
-import pipelines.Transformer
+import workflow.Transformer
 import utils.ChannelMajorArrayVectorizedImage
 import utils.Image
 import utils.ImageUtils

src/main/scala/nodes/images/LabeledImageExtractors.scala

@@ -1,7 +1,7 @@
 package nodes.images
 
-import pipelines._
 import utils.{MultiLabeledImage, Image, LabeledImage}
+import workflow.Transformer
 
 /**
  * Extracts a label from a labeled image.

src/main/scala/nodes/images/PixelScaler.scala

@@ -1,6 +1,6 @@
 package nodes.images
 
-import pipelines.Transformer
+import workflow.Transformer
 import utils.{ImageUtils, Image}
 
 

src/main/scala/nodes/images/Pooler.scala

@@ -3,6 +3,7 @@ package nodes.images
 import breeze.linalg.DenseVector
 import pipelines._
 import utils.{ImageMetadata, ChannelMajorArrayVectorizedImage, Image}
+import workflow.Transformer
 
 /**
  * This node takes an image and performs pooling on regions of the image.

src/main/scala/nodes/images/SIFTExtractor.scala

@@ -1,7 +1,7 @@
 package nodes.images
 
 import breeze.linalg.DenseMatrix
-import pipelines.Transformer
+import workflow.Transformer
 import utils.Image
 
 /**

src/main/scala/nodes/images/SymmetricRectifier.scala

@@ -2,6 +2,7 @@ package nodes.images
 
 import pipelines._
 import utils.{ChannelMajorArrayVectorizedImage, Image}
+import workflow.Transformer
 
 case class SymmetricRectifier(maxVal: Double = 0.0, alpha: Double = 0.0)
   extends Transformer[Image, Image] {

src/main/scala/nodes/images/Windower.scala

@@ -2,7 +2,7 @@ package nodes.images
 
 import breeze.linalg.DenseVector
 import org.apache.spark.rdd.RDD
-import pipelines.{FunctionNode, Transformer}
+import pipelines.FunctionNode
 import utils.{ImageMetadata, ChannelMajorArrayVectorizedImage, Image}
 
 

src/main/scala/nodes/learning/BlockLinearMapper.scala

@@ -5,8 +5,8 @@ import edu.berkeley.cs.amplab.mlmatrix.{RowPartition, NormalEquations, BlockCoor
 import nodes.stats.{StandardScalerModel, StandardScaler}
 import org.apache.spark.rdd.RDD
 import nodes.util.{VectorSplitter, Identity}
-import pipelines.{Transformer, LabelEstimator}
 import utils.{MatrixUtils, Stats}
+import workflow.{Transformer, LabelEstimator}
 
 
 /**

src/main/scala/nodes/learning/BlockWeightedLeastSquares.scala

@@ -1,6 +1,7 @@
 package nodes.learning
 
 import nodes.util.VectorSplitter
+import workflow.LabelEstimator
 
 import scala.collection.mutable.ArrayBuffer
 
@@ -16,7 +17,7 @@ import edu.berkeley.cs.amplab.mlmatrix.{RowPartition, NormalEquations, BlockCoor
 import edu.berkeley.cs.amplab.mlmatrix.util.{Utils => MLMatrixUtils}
 
 import nodes.stats.StandardScaler
-import pipelines.{Transformer, LabelEstimator, Logging}
+import pipelines.Logging
 import utils.{MatrixUtils, Stats}
 
 // Utility class that holds statistics related to each block we solve

src/main/scala/nodes/learning/GaussianMixtureModel.scala

@@ -7,6 +7,7 @@ import org.apache.spark.rdd.RDD
 import pipelines._
 import utils.MatrixUtils
 import utils.external.EncEval
+import workflow.{Transformer, Estimator}
 
 /**
  * A Mixture of Gaussians, usually computed via some clustering process.

src/main/scala/nodes/learning/LinearDiscriminantAnalysis.scala

@@ -3,8 +3,8 @@ package nodes.learning
 import breeze.linalg._
 import breeze.stats._
 import org.apache.spark.rdd.RDD
-import pipelines.LabelEstimator
 import utils.MatrixUtils
+import workflow.LabelEstimator
 
 /**
  * An Estimator that fits Linear Discriminant Analysis (currently not calculated in a distributed fashion),

src/main/scala/nodes/learning/LinearMapper.scala

@@ -2,11 +2,10 @@ package nodes.learning
 
 import breeze.linalg._
 import edu.berkeley.cs.amplab.mlmatrix.{NormalEquations, RowPartitionedMatrix}
-import nodes.stats.{StandardScalerModel, StandardScaler}
+import nodes.stats.{StandardScaler, StandardScalerModel}
 import org.apache.spark.rdd.RDD
-
-import pipelines.{LabelEstimator, Transformer}
 import utils.MatrixUtils
+import workflow.{LabelEstimator, Transformer}
 
 /**
  * Computes A * x + b i.e. a linear map of data using a trained model.

src/main/scala/nodes/learning/NaiveBayesModel.scala

@@ -1,12 +1,11 @@
 package nodes.learning
 
-import breeze.linalg.{argmax, DenseMatrix, DenseVector, Vector}
+import breeze.linalg.{DenseMatrix, DenseVector, Vector}
 import org.apache.spark.mllib.classification.NaiveBayes
 import org.apache.spark.mllib.regression.LabeledPoint
 import org.apache.spark.rdd.RDD
-import pipelines.LabelEstimator
-import pipelines.Transformer
 import utils.MLlibUtils.breezeVectorToMLlib
+import workflow.{Transformer, LabelEstimator}
 
 import scala.reflect.ClassTag
 

src/main/scala/nodes/learning/PCA.scala

@@ -7,6 +7,7 @@ import com.github.fommil.netlib.LAPACK.{getInstance => lapack}
 import org.apache.spark.rdd.RDD
 import org.netlib.util.intW
 import pipelines._
+import workflow.{Transformer, Estimator}
 
 
 /**

src/main/scala/nodes/learning/ZCAWhitener.scala

@@ -7,6 +7,7 @@ import com.github.fommil.netlib.LAPACK.{getInstance => lapack}
 import org.apache.spark.rdd.RDD
 import org.netlib.util.intW
 import pipelines._
+import workflow.{Transformer, Estimator}
 
 class ZCAWhitener(val whitener: DenseMatrix[Double], val means: DenseVector[Double])
   extends Transformer[DenseMatrix[Double],DenseMatrix[Double]] {

src/main/scala/nodes/nlp/CoreNLPFeatureExtractor.scala

@@ -3,7 +3,7 @@ package nodes.nlp
 import edu.arizona.sista.processors.Processor
 import edu.arizona.sista.processors.fastnlp.FastNLPProcessor
 import org.apache.spark.rdd.RDD
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * Transformer that uses CoreNLP to (in order):

src/main/scala/nodes/nlp/StringUtils.scala

@@ -3,7 +3,7 @@ package nodes.nlp
 import java.util.Locale
 
 import org.apache.spark.rdd.RDD
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * Transformer that tokenizes a String into a Seq[String] by splitting on a regular expression.

src/main/scala/nodes/nlp/StupidBackoff.scala

@@ -1,12 +1,12 @@
 package nodes.nlp
 
-import pipelines.{Transformer, Estimator}
-
 import org.apache.spark.Partitioner
 import org.apache.spark.rdd.RDD
 
 import java.util.{HashMap => JHashMap}
 
+import workflow.{Transformer, Estimator}
+
 import scala.annotation.tailrec
 import scala.collection.JavaConverters._
 import scala.reflect.ClassTag

src/main/scala/nodes/nlp/WordFrequencyEncoder.scala

@@ -1,9 +1,8 @@
 package nodes.nlp
 
 import org.apache.spark.broadcast.Broadcast
-import pipelines.{Estimator, Transformer}
-
 import org.apache.spark.rdd.RDD
+import workflow.{Estimator, Transformer}
 
 object WordFrequencyEncoder extends Estimator[Seq[String], Seq[Int]] {
   private[this] def makeUnigrams(data: RDD[Seq[String]]) =

src/main/scala/nodes/nlp/ngrams.scala

@@ -1,11 +1,13 @@
 package nodes.nlp
 
-import pipelines.{FunctionNode, Transformer}
+import pipelines.FunctionNode
 
 import org.apache.spark.rdd.RDD
 
 import java.util.{HashMap => JHashMap}
 
+import workflow.Transformer
+
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
 import scala.reflect.ClassTag

src/main/scala/nodes/stats/CosineRandomFeatures.scala

@@ -6,6 +6,7 @@ import breeze.stats.distributions.Rand
 import org.apache.spark.rdd.RDD
 import pipelines._
 import utils.MatrixUtils
+import workflow.Transformer
 
 /**
  * Transformer that extracts random cosine features from a feature vector

src/main/scala/nodes/stats/LinearRectifier.scala

@@ -2,6 +2,7 @@ package nodes.stats
 
 import breeze.linalg.DenseVector
 import pipelines._
+import workflow.Transformer
 
 /**
  * This transformer applies a Linear Rectifier,

src/main/scala/nodes/stats/NormalizeRows.scala

@@ -2,7 +2,7 @@ package nodes.stats
 
 import breeze.linalg.{max, sum, DenseVector}
 import breeze.numerics._
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * Divides each row by the max of its two-norm and 2.2e-16.

src/main/scala/nodes/stats/PaddedFFT.scala

@@ -2,15 +2,15 @@ package nodes.stats
 
 import breeze.linalg.DenseVector
 import breeze.math.Complex
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * This transformer pads input vectors to the nearest power of two,
  * then returns the real values of the first half of the fourier transform on the padded vectors.
  *
  * Goes from vectors of size n to vectors of size nextPositivePowerOfTwo(n)/2
  */
-object PaddedFFT extends Transformer[DenseVector[Double], DenseVector[Double]] {
+case class PaddedFFT() extends Transformer[DenseVector[Double], DenseVector[Double]] {
   override def apply(in: DenseVector[Double]): DenseVector[Double] = {
     val paddedSize = nextPositivePowerOfTwo(in.length)
     val fft: DenseVector[Complex] = breeze.signal.fourierTr(in.padTo(paddedSize, 0.0).toDenseVector)

src/main/scala/nodes/stats/RandomSignNode.scala

@@ -2,7 +2,7 @@ package nodes.stats
 
 import breeze.linalg._
 import breeze.stats.distributions._
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  *  A node that takes in DenseVector[Double] and randomly flips

src/main/scala/nodes/stats/Sampling.scala

@@ -2,7 +2,7 @@ package nodes.stats
 
 import breeze.linalg.{DenseVector, DenseMatrix}
 import org.apache.spark.rdd.RDD
-import pipelines.{FunctionNode, Transformer}
+import pipelines.FunctionNode
 
 /**
  * Given a collection of Dense Matrices, this will generate a sample of `numSamples` columns from the entire set.

src/main/scala/nodes/stats/SignedHellingerMapper.scala

@@ -2,7 +2,7 @@ package nodes.stats
 
 import breeze.linalg.{DenseVector, DenseMatrix}
 import breeze.numerics._
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  *  Apply power normalization: z <- sign(z)|z|^{\rho}

src/main/scala/nodes/stats/StandardScaler.scala

@@ -4,8 +4,8 @@ import breeze.linalg.DenseVector
 import breeze.numerics.sqrt
 import org.apache.spark.mllib.stat.MultivariateOnlineSummarizer
 import org.apache.spark.rdd.RDD
-import pipelines.{Estimator, Transformer}
 import utils.MLlibUtils
+import workflow.{Transformer, Estimator}
 
 /**
  * Represents a StandardScaler model that can transform dense vectors.

src/main/scala/nodes/stats/TermFrequency.scala

@@ -1,6 +1,6 @@
 package nodes.stats
 
-import pipelines.Transformer
+import workflow.Transformer
 
 /**
  * Transformer that maps a Seq[Any] of objects to a Seq[(Any, Double)] of (unique object, weighting_scheme(tf)),

src/main/scala/nodes/util/AllSparseFeatures.scala

@@ -2,7 +2,7 @@ package nodes.util
 
 import breeze.linalg.SparseVector
 import org.apache.spark.rdd.RDD
-import pipelines.Estimator
+import workflow.Estimator
 
 import scala.reflect.ClassTag
 

src/main/scala/nodes/util/Cacher.scala

@@ -1,7 +1,8 @@
 package nodes.util
 
 import org.apache.spark.rdd.RDD
-import pipelines.{Logging, Transformer}
+import pipelines.Logging
+import workflow.Transformer
 
 import scala.reflect.ClassTag