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