The error is gone. Thanks! Cheers, Max
On Wed, Nov 12, 2014 at 11:41 PM, Stephan Ewen <[email protected]> wrote: > The current master contains a fix for the incorrectly identified nested > iteration bug. > > Please let us know if it fixes your problem! > > Greetings, > Stephan > > > On Wed, Nov 12, 2014 at 5:19 PM, Stephan Ewen <[email protected]> wrote: > >> Talking at the meetup sounds good! >> >> On Wed, Nov 12, 2014 at 5:15 PM, Maximilian Alber < >> [email protected]> wrote: >> >>> Ok. With for loop style you intend a loop with a fixed range? >>> In my case I would have a delta-iteration inside a bulk-iteration. I >>> guess wouldn't be "roll-out-able"? >>> >>> Btw is there any intention to allow bulk-style iterations on several >>> datasets "concurrently"? >>> >>> Maybe we could discuss my problem next week at the meetup? >>> >>> Thank you for the offer, but I'm in the middle of thesis, thus I don't >>> have time for it. >>> >>> Cheers, >>> Max >>> >>> On Wed, Nov 12, 2014 at 4:59 PM, Stephan Ewen <[email protected]> wrote: >>> >>>> We are not planning to add closed-loop nested iterations in the near >>>> future. That is a bit of an effort and so far, and I think no one can pick >>>> that up very soon. >>>> >>>> We will be supporting roll-out iterations (for loop style) much more >>>> efficiently soon. There is no reason why you could not nest two for-loops. >>>> However, those are only bulk-style, not delta-iteration style. >>>> >>>> If you would like to contribute iteration nesting, I could help you to >>>> get started. >>>> >>>> Greetings, >>>> Stephan >>>> >>>> >>>> On Wed, Nov 12, 2014 at 4:47 PM, Maximilian Alber < >>>> [email protected]> wrote: >>>> >>>>> Oh sorry, I just read the bug title. So my questions is when you are >>>>> planning to add nested iterations? >>>>> >>>>> Cheers, >>>>> Max >>>>> >>>>> On Wed, Nov 12, 2014 at 4:45 PM, Maximilian Alber < >>>>> [email protected]> wrote: >>>>> >>>>>> Ok, thanks. >>>>>> >>>>>> But the bug causes that it Flink "sees" a nested iteration where none >>>>>> is? >>>>>> Or is it a bug that nested are not supported? If not when you plan to >>>>>> add this feature? >>>>>> Because I need nested iterations for my algorithm, so it would be >>>>>> nice to know when I can expect them. >>>>>> >>>>>> Cheers, >>>>>> Max >>>>>> >>>>>> On Wed, Nov 12, 2014 at 4:21 PM, Stephan Ewen <[email protected]> >>>>>> wrote: >>>>>> >>>>>>> I found the cause of the bug and have opened a JIRA to track it. >>>>>>> >>>>>>> https://issues.apache.org/jira/browse/FLINK-1235 >>>>>>> >>>>>>> You can watch that one to keep updated. >>>>>>> >>>>>>> Stephan >>>>>>> >>>>>>> >>>>>>> On Wed, Nov 12, 2014 at 2:48 PM, Stephan Ewen <[email protected]> >>>>>>> wrote: >>>>>>> >>>>>>>> Hi! >>>>>>>> >>>>>>>> I am looking into it right now... >>>>>>>> >>>>>>>> Stephan >>>>>>>> >>>>>>>> >>>>>>>> On Tue, Nov 11, 2014 at 2:09 PM, Maximilian Alber < >>>>>>>> [email protected]> wrote: >>>>>>>> >>>>>>>>> Hi Stephan, >>>>>>>>> >>>>>>>>> you already had time to investigate this issue? >>>>>>>>> >>>>>>>>> Cheers, >>>>>>>>> Max >>>>>>>>> >>>>>>>>> On Tue, Oct 21, 2014 at 2:03 PM, Stephan Ewen <[email protected]> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> 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 >>>>>>>>>>> } >>>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >>> >> >
