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 >>> >>> >> >
