Snapshot a pipe in the REPL

**Do not merge yet: feedback needed, work in progress.**

Fixes #879, could be in the territory of #889.

This pull request attempts to address the issue with not being able to run sub-flows in the repl. Rather than resetting the implicit `flowDef` (which disallows re-running pipes -- #879), this adds the ability to run a "sub-flow" for a given TypedPipe that includes only the upstream pipes and sources needed to compute that tail pipe.

This is accessible in the REPL via a couple new enrichments to TypedPipe:
- `.snapshot`: immediately runs and writes the pipe to a temporary SequenceFile, returning a TypedPipe with the SequenceFile as a source.
- `.save(dest)`: same as snapshot but with a user-provided Sink (kinda like shorthand for `.write(dest).run`), also returning a TypedPipe with the saved file as a new Source.

In my current way of thinking, snapshots allow one to inspect intermediate steps of a job as they build it:

~~~scala
> val input = TypedPipe.from(TextLine("tutorial/data/hello.txt"))
> val wordsPipe: TypedPipe[String] = input.flatMap(_.split("\\s+"))
> val s1: TypedPipe[String] = wordsPipe.snapshot
~~~

One then has a choice, when continuing to work on a job, to use the snapshot pipe, if they wish to "memoize" the intermediate work, or use the original pipe, which will cause the flow to be re-evaluated.

Before continuing to work on this, I wanted to elicit some feedback on the intended direction. Overall, I'm hoping to come up with behavior that is least surprising to users of Scalding, but also cater to users coming from other ad-hoc analytics backgrounds (i.e. Pig, Spark)

1. Part of using snapshots should be good ways to actually *view* the contents of the snapshot. I would like to allow a `toList` kind of behavior that would read the contents into memory. This could either be implicit (`.toList` calls `.snapshot` internally), or `.snapshot` could return a new kind of TypedPipe that allowed `.toList` to be called on it. (#709 may be related -- I believe this "sub-flow" ability may fix one of the outstanding issues there)
  - There could also be a Pig-like `dump` call added to snapshots that would just print the contents to stdout (though `.toList` pretty much subsumes this, as you can print a list to stdout)

2. What should the semantics of `run` be in the context of this change? Currently it is broken because all the snapshots add to the implicit flowDef, causing `run` on the whole flow to balk at all the unconnected sources and sinks. Should `run` be fixed by just discarding the extraneous sources and sinks created in the REPL? Or should `run` just go away, in favor of always calling something like `save` to commit the contents of a pipe to a single Sink?

3. Should `snapshot` ultimately be completely hidden from the user? This iteration allows an explicit choice to be made between building off of a snapshot of intermediate results or re-running the whole flow. An alternative would be to only create snapshots when people request to see the contents (via `toList`), and then swap out the memoized pipe for the original so the user gets consistent results. This leads to questions about whether memoized pipes should ever be invalidated when the original contents have changed (which sounds really daunting).

Thoughts?
By the way, I know this patch is a lot of work, but in fact I think it will
turn out to be one of the best contributions in quite a while when done.

I think this can actually fix a design flaw in scalding: that using pipe
and TypedPipe has side effects (mutating FlowDef) that are not easy to
reason about.

Especially with TypedPipe, when this code is all done, even outside of the
repl, we could use it to make TypedPipe referentially transparent, as it
should have been all along.

The trick is: keep a local FlowDef when we make a TypedPipe, and at write
or toPipe, we merge into the implicit one in that scope.

I think this can keep TypedPipe referentially transparent.

On Friday, June 27, 2014, Brandon Holt <notifications@github.com> wrote:

> In scalding-core/src/main/scala/com/twitter/scalding/RichFlowDef.scala:
>
> > +
> > +  /**
> > +   * New flow def with only sources upstream from tails.
> > +   */
> > +  def withoutUnusedSources: FlowDef = {
> > +    import collection.JavaConverters._
> > +
> > +    // find all heads reachable from the tails (as a set of names)
> > +    val heads = fd.getTails.asScala
> > +      .flatMap(_.getHeads).map(_.getName).toSet
> > +    // add taps associated with heads to localFlow
> > +    val filteredSources = fd.getSources.asScala.filterKeys(src => heads.contains(src)).asJava
> > +
> > +    val newFd = fd.copy
> > +    newFd.getSources.clear()
> > +    newFd.addSources(filteredSources)
>
> Ah, I see, forceToDisk.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/twitter/scalding/pull/918/files#r14320751>.
>


-- 
Oscar Boykin :: @posco :: http://twitter.com/posco