Hi Stefan, yes, delta iterations are not very intuitive right now. Your join with the solution set from the previous step to get at the data. The result of that join must then be the result of the iteration and is used to update the solution set before the next iteration.
We are working on making iterations more intuitive for our 0.7 release. Then we would have explicit methods to update the solution instead of the join. Cheers, Aljoscha On Thu, Aug 21, 2014 at 3:51 PM, Stefan Bunk <[email protected]> wrote: > Hi Aljoscha, > > thanks for the reply. Yes, I could get it working. > > Why do I actually have to join with the solution set? In my case, this is > a no-op. The data set is the same before and after the join. I am doing it > only, because I have to. > Am I using it wrong? > > Cheers > Stefan > > > > On Thu, Aug 14, 2014 at 12:46 PM, Aljoscha Krettek <[email protected]> > wrote: > >> Hi Stefan, >> sorry for the late reply but I finally found whats wrong: In Delta >> Iterations, the key you specify in the iterateWithDelta method and in the >> join with the solution set has to be the same. In your join you are using: >> where { cl => (cl.to, cl.origTo) } and in the iterateWithDelta call you >> are using { cl => (cl.from, cl.origTo) }. Is it possible for your algorithm >> to use the same key in both positions? >> >> Cheers, >> Aljoscha >> >> >> On Fri, Aug 1, 2014 at 8:27 AM, Aljoscha Krettek <[email protected]> >> wrote: >> >>> Hi Stefan, >>> sorry for the delay, I'm on vacation right now. :D >>> >>> I just want you to know I'm looking into this. >>> >>> Aljoscha >>> >>> >>> On Thu, Jul 31, 2014 at 4:08 PM, Stefan Bunk <[email protected] >>> > wrote: >>> >>>> Hello, >>>> >>>> I built a plan using Delta-Iteration, however, at runtime, when the >>>> plan is executed, I get the following error: >>>> >>>> Exception in thread "main" >>>> eu.stratosphere.compiler.CompilerPostPassException: Could not set up >>>> runtime strategy for input channel to node ''. Missing type information for >>>> field 0 >>>> I have no idea, where to look now, can anyone help? >>>> >>>> In case it is needed, please find below a description of the problem >>>> and the code, as well as the stacktrace: >>>> Usecase: >>>> I have a list of edges like this (lets call them base links): >>>> a --> b >>>> e --> f >>>> >>>> And a set of (lets call them additional links): >>>> b --> c >>>> c --> d >>>> f --> g >>>> >>>> I want to transitively follow the base links using the additional >>>> links, so that I finally have: >>>> a --> d (over b and c) >>>> e --> g >>>> >>>> That's the problem, here's my delta iteration code (Not sure, what's >>>> best practice for the code - append to the email or attach?): >>>> >>>> override def getPlan(args: String*): Plan = { >>>> // store the orig destination as well, because we need it for the key >>>> in the delta iteration >>>> case class BaseLink(from: String, origTo: String, to: String) >>>> case class AdditionalLink(from: String, to: String) >>>> >>>> // load datasets >>>> val baseLinks = DataSource(baseLinkPath, >>>> CsvInputFormat[BaseLink]("\n", '\t')) >>>> val additionalLinks = DataSource(additionalLinkPath, >>>> CsvInputFormat[AdditionalLink]("\n", '\t')) >>>> >>>> def iterate(s: DataSet[BaseLink], ws: DataSet[BaseLink]) = { >>>> // first, join base links "to" with additional links "from", to do >>>> one resolving step >>>> val resolvedLinks = ws.join(additionalLinks) >>>> .where { case BaseLink(from, to, orig) => to } >>>> .isEqualTo { case AdditionalLink(from, to) => from } >>>> .map { case (baseLink, additionalLink) => >>>> // If we had base link a --> b, and additional link b --> c, we now >>>> have a --> c >>>> BaseLink(baseLink.from, baseLink.origTo, additionalLink.to) >>>> } >>>> // now, join with the solution set to build the solution delta and >>>> the next working set >>>> val nextWs = s.join(resolvedLinks) >>>> .where { cl => (cl.to, cl.origTo) } >>>> .isEqualTo { cl => (cl.to, cl.origTo) } >>>> .map { (orig, resolved) => >>>> orig >>>> } >>>> // nextWs is solution delta as well >>>> (nextWs, nextWs) >>>> } >>>> // start iteration, key is combination of from and orig to, e.g. a >>>> --> b >>>> val resolvedLinks = baseLinks >>>> .iterateWithDelta(baseLinks, { cl => (cl.from, cl.origTo) }, iterate, >>>> 10) >>>> .map { cl => (cl.from, cl.to) } >>>> >>>> val output = resolvedLinks.write(resolvedLinksPath, >>>> CsvOutputFormat[(String, String)]("\n", "\t")) >>>> val plan = new ScalaPlan(Seq(output)) >>>> plan >>>> } >>>> >>>> Here's the whole stacktrace: >>>> Exception in thread "main" >>>> eu.stratosphere.compiler.CompilerPostPassException: Could not set up >>>> runtime strategy for input channel to node ''. Missing type information for >>>> field 0 >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:316) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:228) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.propagateToChannel(GenericFlatTypePostPass.java:499) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:313) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.propagateToChannel(GenericFlatTypePostPass.java:499) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.traverse(GenericFlatTypePostPass.java:91) >>>> at >>>> eu.stratosphere.compiler.postpass.GenericFlatTypePostPass.postPass(GenericFlatTypePostPass.java:66) >>>> at eu.stratosphere.api.scala.ScalaPostPass.postPass(ScalaPlan.scala:77) >>>> at >>>> eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:744) >>>> at eu.stratosphere.compiler.PactCompiler.compile(PactCompiler.java:553) >>>> at >>>> eu.stratosphere.client.LocalExecutor.optimizerPlanAsJSON(LocalExecutor.java:320) >>>> // error happens as well, when I directly call LocalExecutor.execute >>>> >>>> Thanks in advance! >>>> Stefan >>>> >>>> >>> >> >
