Named steps and other goodies

.gitignore

@@ -4,3 +4,5 @@
 .settings
 bin/
 target/
+logs/
+pipeline-output/

build.sbt

@@ -2,11 +2,21 @@ import Dependencies._
 
 import ReleaseKeys._
 
-val pipeline = Project(
-  id = "allenai-pipeline",
-  base = file(".")
+val core = Project(
+  id = "core",
+  base = file("core")
 )
 
+val s3 = Project(
+  id = "s3",
+  base = file("s3")
+).dependsOn(core)
+
+val spark = Project(
+  id = "spark",
+  base = file("spark")
+).dependsOn(core, s3)
+
 organization := "org.allenai"
 crossScalaVersions := Seq("2.11.5")
 scalaVersion <<= crossScalaVersions { (vs: Seq[String]) => vs.head }
@@ -33,12 +43,3 @@ enablePlugins(LibraryPlugin)
 PublishTo.ai2Public
 
 dependencyOverrides += "org.scala-lang" % "scala-reflect" % "2.11.5"
-
-libraryDependencies ++= Seq(
-  sprayJson,
-  awsJavaSdk,
-  commonsIO,
-  ai2Common,
-  allenAiTestkit % "test",
-  scalaReflection
-)

core/build.sbt

@@ -0,0 +1,15 @@
+import Dependencies._
+
+name := "pipeline-core"
+organization := "org.allenai"
+
+StylePlugin.enableLineLimit := false
+
+dependencyOverrides += "org.scala-lang" % "scala-reflect" % "2.11.5"
+libraryDependencies ++= Seq(
+  sprayJson,
+  commonsIO,
+  ai2Common,
+  allenAiTestkit % "test",
+  scalaReflection
+)

src/main/resources/org/allenai/pipeline/pipelineSummary.html → core/src/main/resources/org/allenai/pipeline/pipelineSummary.html

@@ -61,7 +61,12 @@
         padding-top: 10px;
       }
 
-      .node rect {
+        .node .data {
+          border-top: 1px solid #2a75a1;
+          padding-top: 10px;
+        }
+
+        .node rect {
         stroke: #2a75a1;
         stroke-width: 2px;
         fill: #e2f0f8;
@@ -275,7 +280,12 @@
         fill: #C38888;
       }
 
-      #outputContainer {
+        .executionInfo {
+         font-size: 20%%;
+         float: right;
+        }
+
+        #outputContainer {
         position: absolute;
         top: 41px;
         left: 41px;
@@ -313,7 +323,7 @@ <h2>Outputs</h2>
 %s
     </div>
     <script src="http://d3js.org/d3.v3.min.js"></script>
-    <script src="http://cpettitt.github.io/project/dagre-d3/v0.4.2/dagre-d3.min.js"></script>
+    <script src="http://cpettitt.github.io/project/dagre-d3/v0.4.6/dagre-d3.min.js"></script>
     <script>
       (function() {
         /**
@@ -337,42 +347,19 @@ <h2>Outputs</h2>
          *
          * @return {string}   The HTML for the step contents.
          */
-        function generateStepContent(label, description, millis, data, links) {
+        function generateStepContent(label, description, execInfo, data, links) {
           var out = "<h2>" + label + "</h2>";
-          if(millis) {
-            var seconds = 1000;
-            var minutes = seconds * 60;
-            var hours = minutes * 60;
-            var days = hours * 24;
-            if (millis / days > 1) {
-              out += (millis / days).toFixed(2) + " days";
-            }
-            else if (millis / hours > 1) {
-              out += (millis / hours).toFixed(2) + " hours";
-            }
-            else if (millis / minutes > 1) {
-              out += (millis / minutes).toFixed(2) + " minutes";
-            }
-            else if (millis / seconds > 1) {
-              out += (millis / seconds).toFixed(2) + " seconds";
-            }
-            else {
-              out += millis.toFixed(0) + " millis";
-            }
-          }
-          else {
-            out += "cached";
-          }
           if(description) {
             out += "<p>" + description + "</p>";
           }
-          if(data && Array.isArray(data)) {
+          if(data && Array.isArray(data) && data.length > 0) {
             out += '<ul class="data">';
             data.forEach(function(d) {
               out += '<li>' + d + '</li>';
             });
             out += '</ul>';
-            if(links) {
+          }
+          if(links && links.length > 0) {
               out += '<ul class="links">';
               links.forEach(function(l) {
                 if(l instanceof Link) {
@@ -381,7 +368,7 @@ <h2>Outputs</h2>
               });
               out += '</ul>';
             }
-          }
+          out += '<span class="executionInfo">' + execInfo + '</span>';
           return out;
         };
 
@@ -395,6 +382,12 @@ <h2>Outputs</h2>
         // our nodes.
 %s
 
+       // Round the corners of the nodes
+        g.nodes().forEach(function(v) {
+          var node = g.node(v);
+          node.rx = node.ry = 10;
+        });
+
         // Add edges to the graph. The first argument is the edge id. Here we use null
         // to indicate that an arbitrary edge id can be assigned automatically. The
         // second argument is the source of the edge. The third argument is the target

src/main/scala/org/allenai/pipeline/Artifact.scala → core/src/main/scala/org/allenai/pipeline/Artifact.scala

@@ -91,7 +91,7 @@ trait StructuredArtifact extends Artifact {
   */
 class ArtifactStreamWriter(out: OutputStream) {
   def write(data: Array[Byte]): Unit = {
-    out.write(data, 0, data.size)
+    out.write(data, 0, data.length)
   }
 
   def write(data: Array[Byte], offset: Int, size: Int): Unit = {

core/src/main/scala/org/allenai/pipeline/ArtifactFactory.scala

@@ -0,0 +1,126 @@
+package org.allenai.pipeline
+
+import scala.reflect.ClassTag
+
+import java.io.File
+import java.net.URI
+
+/** Creates an Artifact from a URL
+  */
+trait ArtifactFactory {
+  /** @param url The location of the Artifact.  The scheme (protocol) is used to determine the
+    *            specific implementation.
+    * @tparam A The type of the Artifact to create.  May be an abstract or concrete type
+    * @return The artifact
+    */
+  def createArtifact[A <: Artifact: ClassTag](url: URI): A
+
+  /** If path is an absolute URL, create an Artifact at that location.
+    * If it is a relative path, create it relative to the given root URL
+    */
+  def createArtifact[A <: Artifact: ClassTag](rootUrl: URI, path: String): A = {
+    val parsed = new URI(path)
+    val url = parsed.getScheme match {
+      case null =>
+        val fullPath = s"${rootUrl.getPath.reverse.dropWhile(_ == '/').reverse}/${parsed.getPath.dropWhile(_ == '/')}"
+        new URI(
+          rootUrl.getScheme,
+          rootUrl.getHost,
+          fullPath,
+          rootUrl.getFragment
+        )
+      case _ => parsed
+    }
+    createArtifact[A](url)
+  }
+}
+
+object ArtifactFactory {
+  def apply(urlHandler: UrlToArtifact, fallbackUrlHandlers: UrlToArtifact*): ArtifactFactory =
+    new ArtifactFactory {
+      val urlHandlerChain =
+        if (fallbackUrlHandlers.isEmpty) {
+          urlHandler
+        } else {
+          UrlToArtifact.chain(urlHandler, fallbackUrlHandlers.head, fallbackUrlHandlers.tail: _*)
+        }
+
+      def createArtifact[A <: Artifact: ClassTag](url: URI): A = {
+        val fn = urlHandlerChain.urlToArtifact[A]
+        val clazz = implicitly[ClassTag[A]].runtimeClass.asInstanceOf[Class[A]]
+        require(fn.isDefinedAt(url), s"Cannot create $clazz from $url")
+        fn(url)
+      }
+    }
+
+}
+
+/** Supports creation of a particular type of Artifact from a URL.
+  * Allows chaining together of different implementations that recognize different input URLs
+  * and support creation of different Artifact types
+  */
+trait UrlToArtifact {
+  /** Return a PartialFunction indicating whether the given Artifact type can be created from an input URL
+    * @tparam A The Artifact type to be created
+    * @return A PartialFunction where isDefined will return true if an Artifact of type A can
+    *         be created from the given URL
+    */
+  def urlToArtifact[A <: Artifact: ClassTag]: PartialFunction[URI, A]
+}
+
+object UrlToArtifact {
+  // Chain together a series of UrlToArtifact instances
+  // The result will be a UrlToArtifact that supports creation of the union of Artifact types and input URLs
+  // that are supported by the individual inputs
+  def chain(first: UrlToArtifact, second: UrlToArtifact, others: UrlToArtifact*) =
+    new UrlToArtifact {
+      override def urlToArtifact[A <: Artifact: ClassTag]: PartialFunction[URI, A] = {
+        var fn = first.urlToArtifact[A] orElse second.urlToArtifact[A]
+        for (o <- others) {
+          fn = fn orElse o.urlToArtifact[A]
+        }
+        fn
+      }
+    }
+
+  object Empty extends UrlToArtifact {
+    def urlToArtifact[A <: Artifact: ClassTag]: PartialFunction[URI, A] =
+      PartialFunction.empty[URI, A]
+  }
+
+}
+
+object CreateCoreArtifacts {
+  // Create a FlatArtifact or StructuredArtifact from an absolute file:// URL
+  val fromFileUrls: UrlToArtifact = new UrlToArtifact {
+    def urlToArtifact[A <: Artifact: ClassTag]: PartialFunction[URI, A] = {
+      val c = implicitly[ClassTag[A]].runtimeClass.asInstanceOf[Class[A]]
+      val fn: PartialFunction[URI, A] = {
+        case url if c.isAssignableFrom(classOf[FileArtifact])
+          && "file" == url.getScheme =>
+          new FileArtifact(new File(url)).asInstanceOf[A]
+        case url if c.isAssignableFrom(classOf[FileArtifact])
+          && null == url.getScheme =>
+          new FileArtifact(new File(url.getPath)).asInstanceOf[A]
+        case url if c.isAssignableFrom(classOf[DirectoryArtifact])
+          && "file" == url.getScheme
+          && new File(url).exists
+          && new File(url).isDirectory =>
+          new DirectoryArtifact(new File(url)).asInstanceOf[A]
+        case url if c.isAssignableFrom(classOf[DirectoryArtifact])
+          && null == url.getScheme
+          && new File(url.getPath).exists
+          && new File(url.getPath).isDirectory =>
+          new DirectoryArtifact(new File(url.getPath)).asInstanceOf[A]
+        case url if c.isAssignableFrom(classOf[ZipFileArtifact])
+          && "file" == url.getScheme =>
+          new ZipFileArtifact(new File(url)).asInstanceOf[A]
+        case url if c.isAssignableFrom(classOf[ZipFileArtifact])
+          && null == url.getScheme =>
+          new ZipFileArtifact(new File(url.getPath)).asInstanceOf[A]
+      }
+      fn
+    }
+  }
+}
+

src/main/scala/org/allenai/pipeline/ArtifactIo.scala → core/src/main/scala/org/allenai/pipeline/ArtifactIo.scala

@@ -9,23 +9,23 @@ import scala.io.{ Codec, Source }
 import scala.reflect.ClassTag
 
 trait ArtifactIo[T, -A <: Artifact]
-  extends SerializeToArtifact[T, A] with DeserializeFromArtifact[T, A]
+  extends Serializer[T, A] with Deserializer[T, A]
 
 /** Interface for defining how to persist a data type.
   *
   * @tparam  T  the type of the data being serialized
-  * @tparam  A  the type of the artifact being written (i.e. FileArtifact)
+  * @tparam  A  the type of the artifact being written (e.g. FileArtifact)
   */
-trait SerializeToArtifact[-T, -A <: Artifact] extends PipelineStep {
+trait Serializer[-T, -A <: Artifact] extends PipelineStep {
   def write(data: T, artifact: A): Unit
 }
 
 /** Interface for defining how to persist a data type.
   *
   * @tparam  T  the type of the data being serialized
-  * @tparam  A  the type of the artifact being read (i.e. FileArtifact)
+  * @tparam  A  the type of the artifact being read (e.g. FileArtifact)
   */
-trait DeserializeFromArtifact[+T, -A <: Artifact] extends PipelineStep {
+trait Deserializer[+T, -A <: Artifact] extends PipelineStep {
   def read(artifact: A): T
 }
 
@@ -54,11 +54,14 @@ class SingletonIo[T: StringSerializable: ClassTag](implicit codec: Codec)
     _.write(implicitly[StringSerializable[T]].toString(data))
   }
 
-  override def stepInfo: PipelineStepInfo =
+  override def stepInfo: PipelineStepInfo = {
+    val className = scala.reflect.classTag[T].runtimeClass.getSimpleName
     super.stepInfo.copy(
-      className = s"SingletonIo[${scala.reflect.classTag[T].runtimeClass.getSimpleName}]",
-      parameters = Map("charSet" -> codec.charSet.toString)
+      className = s"ReadObject[$className]",
+      parameters = Map("charSet" -> codec.charSet.toString),
+      description = Some(s"Read [$className] into memory")
     )
+  }
 }
 
 object SingletonIo {
@@ -82,11 +85,14 @@ class LineCollectionIo[T: StringSerializable: ClassTag](implicit codec: Codec)
   override def write(data: Iterable[T], artifact: FlatArtifact): Unit =
     delegate.write(data.iterator, artifact)
 
-  override def stepInfo: PipelineStepInfo =
+  override def stepInfo: PipelineStepInfo = {
+    val className = scala.reflect.classTag[T].runtimeClass.getSimpleName
     super.stepInfo.copy(
-      className = s"LineCollectionIo[${scala.reflect.classTag[T].runtimeClass.getSimpleName}]",
-      parameters = Map("charSet" -> codec.charSet.toString)
+      className = s"ReadCollection[$className]",
+      parameters = Map("charSet" -> codec.charSet.toString),
+      description = Some(s"Read collection of [$className] into memory")
     )
+  }
 
 }
 
@@ -124,12 +130,15 @@ class LineIteratorIo[T: StringSerializable: ClassTag](implicit codec: Codec)
     }
   }
 
-  override def stepInfo: PipelineStepInfo =
+  override def stepInfo: PipelineStepInfo = {
+    val className = scala.reflect.classTag[T].runtimeClass.getSimpleName
     super.stepInfo.copy(
       className =
-      s"LineIteratorIo[${scala.reflect.classTag[T].runtimeClass.getSimpleName}]",
-      parameters = Map("charSet" -> codec.charSet.toString)
+      s"ReadIterator[$className]",
+      parameters = Map("charSet" -> codec.charSet.toString),
+      description = Some(s"Stream iterator of [$className]")
     )
+  }
 }
 
 object LineIteratorIo {

src/main/scala/org/allenai/pipeline/CodeInfo.scala → core/src/main/scala/org/allenai/pipeline/CodeInfo.scala

@@ -34,7 +34,7 @@ object Ai2CodeInfo {
         // We have to guess which remote will have the commit in it
         val useRemote = remotes.size match {
           // If there is only one remote, use it
-          case 1 => remotes(0)
+          case 1 => remotes.head
           // People shouldn't push directly to the upstream allenai repo.  Instead the upstream
           // repo gets updated via a pull request, which will have a different commit sha
           // Use the first non-allenai repo found in the list, which will typically be