Hey! Clearly, this looks like a bug. Let me investigate that and get back at you later...
Greetings, Stephan On Tue, Oct 21, 2014 at 1:16 PM, Maximilian Alber < [email protected]> wrote: > Hi Flinksters! > > First some good news: the cumsum code from the last issue works now > correctly and is tested. > > Bad news (at least for me): I just run into this (for the error and code > see below). You have a road map when this feature will be available? > Regardless of the rest, I would need it in the near future. > > So far so good. But I wonder where this nested iteration should be. At > least I do not see them... I have an iteration and inside a lot of > filters/maps/etc. but not another iteration. > > Cheers, > Max > > Error: > > org.apache.flink.compiler.CompilerException: An error occurred while > translating the optimized plan to a nephele JobGraph: An error occurred > while translating the optimized plan to a nephele JobGraph: Nested > Iterations are not possible at the moment! > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:543) > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:95) > at > org.apache.flink.compiler.plan.DualInputPlanNode.accept(DualInputPlanNode.java:170) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.OptimizedPlan.accept(OptimizedPlan.java:165) > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.compileJobGraph(NepheleJobGraphGenerator.java:163) > at org.apache.flink.client.program.Client.getJobGraph(Client.java:218) > at org.apache.flink.client.program.Client.run(Client.java:290) > at org.apache.flink.client.program.Client.run(Client.java:285) > at org.apache.flink.client.program.Client.run(Client.java:230) > at org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:347) > at org.apache.flink.client.CliFrontend.run(CliFrontend.java:334) > at > org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1001) > at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1025) > Caused by: org.apache.flink.compiler.CompilerException: An error occurred > while translating the optimized plan to a nephele JobGraph: Nested > Iterations are not possible at the moment! > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:543) > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:95) > at > org.apache.flink.compiler.plan.DualInputPlanNode.accept(DualInputPlanNode.java:170) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:199) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:199) > at > org.apache.flink.compiler.plan.DualInputPlanNode.accept(DualInputPlanNode.java:163) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.DualInputPlanNode.accept(DualInputPlanNode.java:163) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.SingleInputPlanNode.accept(SingleInputPlanNode.java:196) > at > org.apache.flink.compiler.plan.WorksetIterationPlanNode.acceptForStepFunction(WorksetIterationPlanNode.java:195) > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:398) > ... 14 more > Caused by: org.apache.flink.compiler.CompilerException: Nested Iterations > are not possible at the moment! > at > org.apache.flink.compiler.plantranslate.NepheleJobGraphGenerator.postVisit(NepheleJobGraphGenerator.java:395) > ... 33 more > > Code: > > def createPlanFirstIteration(env: ExecutionEnvironment) = { > val X = env readTextFile config.xFile map > {Vector.parseFromString(config.dimensions, _)} > val residual = env readTextFile config.yFile map > {Vector.parseFromString(_)} > val randoms = env readTextFile config.randomFile map > {Vector.parseFromString(_)} > val widthCandidates = env readTextFile config.widthCandidatesFile map > {Vector.parseFromString(config.dimensions, _)} > > val center = calcCenter(env, X, residual, randoms, 0) > > val x = calcWidthHeight(env, X, residual, widthCandidates, center) > > x map { _ toString } writeAsText config.outFile > } > > def calcCenter(env: ExecutionEnvironment, X: DataSet[Vector], residual: > DataSet[Vector], randoms: DataSet[Vector], iteration: Int): DataSet[Vector] > = { > val residual_2 = residual * residual > val ys = (residual_2 sumV) * (randoms filter {_.id == iteration} > neutralize) > > val emptyDataSet = env.fromCollection[Vector](Seq()) > val sumVector = env.fromCollection(Seq(Vector.zeros(1))) > val cumSum = emptyDataSet.iterateDelta(sumVector union residual_2, > config.N+1, Array("id")) { > (solutionset, workset) => > val current = workset filter (new RichFilterFunction[Vector]{ > def filter(x: Vector) = x.id == > (getIterationRuntimeContext.getSuperstepNumber-1) > }) > val old_sum = workset filter {_.id == -1} > val sum = VectorDataSet.add(old_sum, current.neutralize()) > > val new_workset = workset filter {_.id != -1} union sum > (sum map (new RichMapFunction[Vector, Vector]{ > def map(x: Vector) = new > Vector(getIterationRuntimeContext.getSuperstepNumber-1, x.values) > }), > new_workset) > } > val index = cumSum.filter(new RichFilterFunction[Vector](){ > var y: Vector = null > override def open(config: Configuration) = { > y = getRuntimeContext.getBroadcastVariable("ys").toList.head > } > def filter(x: Vector) = x.values(0) < y.values(0) > }).withBroadcastSet(ys, "ys") map {x: Vector => Tuple1(1)} sum 0 > > val center = X.filter(new RichFilterFunction[Vector](){ > var index: Int = -1 > override def open(config: Configuration) = { > val x: Tuple1[Int] = > getRuntimeContext.getBroadcastVariable("index").toList.head > index = x._1 > } > def filter(x: Vector) = x.id == index > }).withBroadcastSet(index, "index") > > center neutralize > } > > def getKernelVector(X: DataSet[Vector], center: DataSet[Vector], width: > DataSet[Vector]): DataSet[Vector] = { > X.map(new RichMapFunction[Vector, Vector]{ > var center: Vector = null > var width: Vector = null > override def open(config: Configuration) = { > center = > getRuntimeContext.getBroadcastVariable("center").toList.head > width = getRuntimeContext.getBroadcastVariable("width").toList.head > } > > def map(x: Vector) = new Vector(x.id, > Array(Math.exp(-((((x-center)*(x-center))/width).values.sum)).toFloat)) > }).withBroadcastSet(center, "center").withBroadcastSet(width, "width") > } > > > def calcWidthHeight(env: ExecutionEnvironment, X: DataSet[Vector], > residual: DataSet[Vector], widthCandidates: DataSet[Vector], center: > DataSet[Vector]): DataSet[Vector] = { > val emptyDataSet = env.fromCollection[Vector](Seq()) > val costs = emptyDataSet.iterateDelta(widthCandidates, > config.NWidthCandidates, Array("id")) { > (solutionset, workset) => > val currentWidth = workset filter (new RichFilterFunction[Vector]{ > def filter(x: Vector) = x.id == > (getIterationRuntimeContext.getSuperstepNumber-1) > }) > > val kernelVector = getKernelVector(X, center, currentWidth) > > val x1 = kernelVector dot residual map {x => x*x} > val x2 = kernelVector dot kernelVector > > val cost = (x1 / x2) neutralize > > > (cost map (new RichMapFunction[Vector, Vector]{ > def map(x: Vector) = new > Vector(getIterationRuntimeContext.getSuperstepNumber-1, x.values) > }), > workset) > } > > // todo: will not work > //val width = costs max(0) > > //val kernelVector = getKernelVector(X, center, width) > > //val x1 = kernelVector dot residual > //val x2 = kernelVector dot kernelVector > //val height = x1 / x2 > costs > } >
