Move toIterableExecution and forceToDiskExecution into Execution #1682

johnynek · 2017-05-06T00:50:02Z

This is a follow up to #1681

Here we have done almost all of the work to make Execution delegate to a Writer that handles all the side effects.

With this we can basically do 100% of having an in-memory or spark backend as long as users don't use certain cascading only features (toPipe, TypedPipe from a Pipe, or Execution from a FlowDef).

We are getting very close to where I want to be for the next published version. I want to make an example in-memory backend just to exercise everything to see if we really can run non-trivial jobs. We can leave a spark or flink backend to a future work.

I would like to remove the Execution from a FlowDef, I doubt it was ever used. I'd like to move toPipe to an implicit enrichment that users opt into if they really need it (which I doubt they would).

johnynek · 2017-05-08T17:57:16Z

@ianoc any time for this one?

johnynek · 2017-05-08T19:27:26Z

ping @benpence

benpence

I think I understand what's going on in each localized scope, but am less familiar with the whole picture here.

Overall, it looks fine. Left a few questions/comments.

benpence · 2017-05-08T19:40:57Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+        cache.writer match {
+          case ar: AsyncFlowDefRunner =>
+            ar.validateAndRun(conf, mode)(result).map { m => ((), Map(m)) }
+          case other =>


Why not also make this part of the interface like in #start() instead of matching? Do we anticipate handling writers differently?

I'm not sure we can easily do this differently with the current interface -- in that FlowDefExecution is tied to cascading. this match would eventually go away with a more cross platform interface I would think. Since we would want to eventually get references to FlowDef out of this file

what Ian said. The goal here is to make hopefully all of the typed API not depend on cascading and just make the cascading backend one implementation. I think no one uses this function anyway (creating an Execution from a FlowDef) and I want to just remove it, but in the mean time, I want to not dig us in any deeper.

If no one at Twitter is using it, I would suggest we remove the function, but keeping it around for 1 version deprecated may work okay too.

We do have a few #fromFn uses in our code, specifically in a couple places where we have FlowDef => Unit functions to accommodate both Job & Execution. Deprecation is definitely a good idea though.

benpence · 2017-05-09T14:38:44Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

-                val summedCounters = (fdCounters :: lCounters).reduce(_ ++ _)
-                (fn(conf, mode), summedCounters)
-            }
+            val bothFutures = failFastZip(otherResult, localFlowDefCountersFuture)


So this localFlowDefCountersFuture is a Future for the stats from finishing the cascading flow?

First note, I didn't change that, I just captured the future into a name so I could use a for comprehension and have it read nicer, in my opinion.

About how this works, I hope the code is somewhat readable, but it is not simple. This is where a lot of the hard work of Execution is happening and it got more complicated over time due to finding and fixing issues with its use at Twitter. The thing we are working very hard to do here is not ever re-run a ToWrite aside from also not re-running ANY execution. The problem is, if things are zipped differently, the same ToWrite might end up in two different Executions. So we look through for all the ToWrites we didn't yet run, we run those in one call to the Writer, and we merge the counter results from the other previous writes (otherwise the counter results are wrong).

Ya I was just a little confused by what was going on here in the old code. I think it makes sense now. I think the concept of "we" and "someoneElse" was not clear to me before.

benpence · 2017-05-09T14:43:47Z

scalding-core/src/main/scala/com/twitter/scalding/typed/TypedPipe.scala

-    Execution.write(writeFn, readFn, filesToDeleteFn)
-  }
+  def forceToDiskExecution: Execution[TypedPipe[T]] =
+    Execution.forceToDisk(this)


This is much easier to read :)

benpence · 2017-05-09T14:44:40Z

...ng-core/src/main/scala/com/twitter/scalding/typed/cascading_backend/AsyncFlowDefRunner.scala

@@ -68,14 +73,51 @@ object AsyncFlowDefRunner {
 }

 /**
+ * This holds an internal thread to run
 * This holds an internal thread to submit run


Duplicate comment

that's a merge issue I think... :/

benpence · 2017-05-09T14:51:29Z

...ng-core/src/main/scala/com/twitter/scalding/typed/cascading_backend/AsyncFlowDefRunner.scala

+      s
+    }
+  private def getState: State =
+    updateState { s => (s, s) }


I find this a little hard to read because of the use of S and s and I anticipate people unfamiliar with this construct will find it confusing. Can we use R and result instead or some more verbose in the naming?

can you be concrete about how you want to change this? These is only one variable here, the state and you are copying that to be the result. Can you write the code you want me to replace it with because I don't follow your comment.

Sure something along the lines of

private def updateState[R](modify: State => (State, R)): R = mutex.synchronized { val (newState, result) = modify(state) state = newState result } private def getState: State = // Expose current state as the result value updateState { state => (state, state) }

although as is it's probably OK.

I'll sweep this change into the next one if that's okay.

ianoc · 2017-05-09T18:29:25Z

...ng-core/src/main/scala/com/twitter/scalding/typed/cascading_backend/AsyncFlowDefRunner.scala

 * This holds an internal thread to submit run
 * a Config, Mode, FlowDef and return a Future holding the
 * JobStats
 */
 class AsyncFlowDefRunner extends Writer { self =>
  import AsyncFlowDefRunner._

-  private[this] val filesToCleanup = mutable.Set[String]()
+  private[this] val mutex = new AnyRef


do we need the mutex rather than self?

personal OCD. I don't like to lock things others can see. I always like to lock a private val.

benpence · 2017-05-09T18:49:14Z

...ng-core/src/main/scala/com/twitter/scalding/typed/cascading_backend/AsyncFlowDefRunner.scala

+      s
+    }
+  private def getState: State =
+    updateState { s => (s, s) }


Sure something along the lines of

private def updateState[R](modify: State => (State, R)): R = mutex.synchronized { val (newState, result) = modify(state) state = newState result } private def getState: State = // Expose current state as the result value updateState { state => (state, state) }

although as is it's probably OK.

benpence · 2017-05-09T19:07:58Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

-                val summedCounters = (fdCounters :: lCounters).reduce(_ ++ _)
-                (fn(conf, mode), summedCounters)
-            }
+            val bothFutures = failFastZip(otherResult, localFlowDefCountersFuture)


Ya I was just a little confused by what was going on here in the old code. I think it makes sense now. I think the concept of "we" and "someoneElse" was not clear to me before.

benpence · 2017-05-09T19:20:32Z

scalding-core/src/main/scala/com/twitter/scalding/Execution.scala

+        cache.writer match {
+          case ar: AsyncFlowDefRunner =>
+            ar.validateAndRun(conf, mode)(result).map { m => ((), Map(m)) }
+          case other =>


We do have a few #fromFn uses in our code, specifically in a couple places where we have FlowDef => Unit functions to accommodate both Job & Execution. Deprecation is definitely a good idea though.

ianoc · 2017-05-09T19:36:46Z

lgtm

…tter#1682) * Pull cascading flowDef thread into cascading_backend * Make ToWrite public * remove unused imports * Move toIterableExecution and forceToDiskExecution into Execution * use Writer API a bit more

johnynek added 5 commits May 3, 2017 11:53

Pull cascading flowDef thread into cascading_backend

45ffc7c

Make ToWrite public

2585650

remove unused imports

ad57580

Move toIterableExecution and forceToDiskExecution into Execution

819a347

merge with develop

8ab4c22

johnynek requested review from benpence and ianoc May 6, 2017 00:50

use Writer API a bit more

0c54eda

benpence reviewed May 9, 2017

View reviewed changes

ianoc reviewed May 9, 2017

View reviewed changes

benpence approved these changes May 9, 2017

View reviewed changes

johnynek merged commit 6166921 into develop May 10, 2017

johnynek deleted the oscar/move-force-to-disk branch May 10, 2017 00:06

johnynek mentioned this pull request Jun 3, 2017

First draft of pure-scalding memory backend #1697

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move toIterableExecution and forceToDiskExecution into Execution #1682

Move toIterableExecution and forceToDiskExecution into Execution #1682

johnynek commented May 6, 2017

johnynek commented May 8, 2017

johnynek commented May 8, 2017

benpence left a comment

benpence May 8, 2017

ianoc May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

benpence May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

benpence May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

johnynek May 10, 2017

benpence May 10, 2017

ianoc May 9, 2017

johnynek May 9, 2017

benpence May 9, 2017

benpence May 9, 2017

benpence May 9, 2017

ianoc commented May 9, 2017

Move toIterableExecution and forceToDiskExecution into Execution #1682

Move toIterableExecution and forceToDiskExecution into Execution #1682

Conversation

johnynek commented May 6, 2017

johnynek commented May 8, 2017

johnynek commented May 8, 2017

benpence left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ianoc commented May 9, 2017