Good night!
I did repro for infinite loop (with optimization) or exception (without): ttim/scalding@627ccce . cc/ @dieu
@johnynek We added a guard into Fill function to fail on (at least some) mutations https://github.com/twitter/scalding/pull/1894/files . Wdyt?
As follow up items we think about next:
- fix algebird's
PriorityQueuemonoid - find out why full disable of this optimization fail a job. I appreciate your thoughts on that, especially on how critical this optimizations to keep scalding properly working on cascading?
Personally I do think about (1) is to create
ImmutablePriorityQueue abstraction which just holds sorted array under the hood. I'm pretty sure it's going to be even more performant in most our use cases , wdyt?
note this is immutable topk: https://github.com/twitter/algebird/blob/develop/algebird-core/src/main/scala/com/twitter/algebird/TopKMonoid.scala#L20 is that not fast enough?
using an immutable sorted heap would be faster...
https://github.com/twitter/algebird/blob/develop/algebird-core/src/main/scala/com/twitter/algebird/mutable/PriorityQueueMonoid.scala#L26 — @johnynek is question for you :)
in this case, many optimizations that cascading or scalding can do will break users since we assume values are immutable once in the TypedPipe.
BTW: here is an immutable Heap which should given even better perf: https://github.com/typelevel/cats-collections/blob/master/core/src/main/scala/cats/collections/Heap.scala
can you update: https://github.com/twitter/scalding/releases
oh, you mean, internally?
if so, congratulations!
Give us week or two to see if we have any problems
Then we can make official release
@dieu @ttim I saw this: https://cloud.google.com/twitter/ would be interesting to think about a scalding backend for beam, perhaps...
I think that can work for the basic types automatically (parquet, sequencefile, tsv/csv/textline... should cover most cases).
@johnynek we found an issue with new scalding - it fails planning if two hashJoins one after another is presented: ttim/scalding@9532d85 . If you have an idea on what's an issue from top of your head it would be good. I'm currently debugging scalding to find out what's wrong
We used to put more forces in place. We may have misported that code
That's a stacktrace:
could not build flow from assembly: [[_pipe_1-0*IterableSour...][com.twitter.scalding.typed.cascading_backend.CascadingBackend$.com$twitter$scalding$typed$cascading_backend$CascadingBackend$$planHashJoin(CascadingBackend.scala:662)] found duplicate field names in joined tuple stream: ['key', 'value', 'key1', 'value1']['key1', 'value1']]
cascading.flow.planner.PlannerException: could not build flow from assembly: [[_pipe_1-0*IterableSour...][com.twitter.scalding.typed.cascading_backend.CascadingBackend$.com$twitter$scalding$typed$cascading_backend$CascadingBackend$$planHashJoin(CascadingBackend.scala:662)] found duplicate field names in joined tuple stream: ['key', 'value', 'key1', 'value1']['key1', 'value1']]
at cascading.flow.planner.FlowPlanner.handleExceptionDuringPlanning(FlowPlanner.java:578)
at cascading.flow.local.planner.LocalPlanner.buildFlow(LocalPlanner.java:108)
at cascading.flow.local.planner.LocalPlanner.buildFlow(LocalPlanner.java:40)
at cascading.flow.FlowConnector.connect(FlowConnector.java:459)
at com.twitter.scalding.ExecutionContext$class.buildFlow(ExecutionContext.scala:95)
at com.twitter.scalding.ExecutionContext$$anon$1.buildFlow(ExecutionContext.scala:210)
at com.twitter.scalding.typed.cascading_backend.AsyncFlowDefRunner$$anon$2.go$1(AsyncFlowDefRunner.scala:172)
at com.twitter.scalding.typed.cascading_backend.AsyncFlowDefRunner$$anon$2.run(AsyncFlowDefRunner.scala:201)
at java.lang.Thread.run(Thread.java:745)
Caused by: cascading.pipe.OperatorException: [_pipe_1-0*IterableSour...][com.twitter.scalding.typed.cascading_backend.CascadingBackend$.com$twitter$scalding$typed$cascading_backend$CascadingBackend$$planHashJoin(CascadingBackend.scala:662)] found duplicate field names in joined tuple stream: ['key', 'value', 'key1', 'value1']['key1', 'value1']
at cascading.pipe.Splice.resolveDeclared(Splice.java:1299)
at cascading.pipe.Splice.outgoingScopeFor(Splice.java:992)
at cascading.flow.planner.ElementGraph.resolveFields(ElementGraph.java:628)
at cascading.flow.planner.ElementGraph.resolveFields(ElementGraph.java:610)
at cascading.flow.local.planner.LocalPlanner.buildFlow(LocalPlanner.java:95)
... 7 more
Caused by: cascading.tuple.TupleException: field name already exists: key1
at cascading.tuple.Fields.copyRetain(Fields.java:1397)
at cascading.tuple.Fields.appendInternal(Fields.java:1266)
at cascading.tuple.Fields.append(Fields.java:1215)
at cascading.pipe.Splice.resolveDeclared(Splice.java:1290)
... 11 more
we use key1 to do the hashjoin, but we don't then do an identity transformation to remove the old key1...
so it is hanging out until there is another barrier (like groupby or join or something)... I think we need to explicitly put an identity transformation after the hashJoin
@ttim this project https://github.com/twitter/scalding/blob/0.17.x/scalding-core/src/main/scala/com/twitter/scalding/typed/HashJoinable.scala#L60 got lost I think...
if you make a PR with that bug, I think by adding that project back: https://github.com/twitter/scalding/blob/develop/scalding-core/src/main/scala/com/twitter/scalding/typed/cascading_backend/CascadingBackend.scala#L670 will fix (If I am right).
thanks!