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