Correct, no changing. DataSets can never be modified, in fact. (Except the solution set in a delta iteration, but there the user is also not explicitly changing a DataSet.)
On Wed, Oct 8, 2014 at 12:56 PM, Maximilian Alber <[email protected]> wrote: > But no changing? > > On Wed, Oct 8, 2014 at 11:39 AM, Aljoscha Krettek <[email protected]> > wrote: >> >> Yes, you can refer to outside datasets in an iteration. >> >> On Wed, Oct 8, 2014 at 10:37 AM, Maximilian Alber >> <[email protected]> wrote: >> > Just got into my mind: it possible to have broadcast sets inside the >> > iteration functions with datasets which are "located" outside of it (via >> > closure)? >> > >> > The basic type of my iteration is that I have a datasets which gets >> > altered >> > and is needed each iterations aka working set, in my case I have also a >> > constant dataset which gets not modified (that messes up the code) and a >> > resulting dataset which is not needed inside the step function. >> > Thus similar to iterate with delta. >> > >> > Cheers, >> > Max >> > >> > >> > On Wed, Oct 8, 2014 at 10:26 AM, Aljoscha Krettek <[email protected]> >> > wrote: >> >> >> >> Hmm, what it really needs is a different kind of iteration primitive. >> >> Basically a bulk iteration where you can output values in each >> >> iteration that get collected. >> >> >> >> On Wed, Oct 8, 2014 at 10:02 AM, Maximilian Alber >> >> <[email protected]> wrote: >> >> > Hi! >> >> > >> >> > Hmm I don't think so. I have two datasets, which I cannot really >> >> > merge >> >> > together. After some thinking this solution was the only I got for >> >> > solving >> >> > my problem: >> >> > I have a DataSet with Vector(in this case just with length one) each >> >> > has >> >> > an >> >> > id and an array with values. Out of that I would like to create the >> >> > prefix >> >> > sums aka the cumulative sums. To do it I need the to keep the dataset >> >> > with >> >> > the vectors and the dataset where I store the sums. >> >> > >> >> > In the Scala version I could use a dataset inside the iteration >> >> > without >> >> > passing as solution or workset just via closures? >> >> > >> >> > Maybe a flag to disable the check would be suitable? >> >> > >> >> > Thanks! >> >> > Cheers, >> >> > Max >> >> > >> >> > On Tue, Oct 7, 2014 at 4:34 PM, Stephan Ewen <[email protected]> >> >> > wrote: >> >> >> >> >> >> Hey! >> >> >> >> >> >> Is the algorithm you are using a delta iteration in fact. If you >> >> >> actually >> >> >> do not use the solution set, can you model it as a bulk-iteration? >> >> >> >> >> >> If you actually need the solution set to accumulate data, we can >> >> >> probably >> >> >> deactivate that check in the compiler. As far as I remember, there >> >> >> is >> >> >> no >> >> >> requirement in the runtime to join with the solution set. The check >> >> >> is >> >> >> meant >> >> >> to help programmers that forgot the join... >> >> >> >> >> >> Greetings, >> >> >> Stephan >> >> >> >> >> >> >> >> >> >> >> >> On Tue, Oct 7, 2014 at 3:13 PM, Maximilian Alber >> >> >> <[email protected]> wrote: >> >> >>> >> >> >>> Hi Flinksters! >> >> >>> >> >> >>> I would like to use iterateDelta function. I don't need the >> >> >>> solution >> >> >>> set >> >> >>> inside the step function, because I generate a different values out >> >> >>> of >> >> >>> the >> >> >>> working set. Unfortunately the compiler of the development version >> >> >>> doesn't >> >> >>> like that. Is there a workaround? >> >> >>> >> >> >>> The code: >> >> >>> >> >> >>> val residual_2a = residual_2 union >> >> >>> env.fromCollection(Seq(Vector.zeros(config.dimensions))) >> >> >>> val emptyDataSet = env.fromCollection[Vector](Seq()) >> >> >>> val cumSum = emptyDataSet.iterateDelta(residual_2a, 1000000, >> >> >>> Array("id")) >> >> >>> { >> >> >>> (solutionset, workset) => >> >> >>> val old_sum = workset filter {_.id == -1} >> >> >>> val current = workset filter (new RichFilterFunction[Vector]{ >> >> >>> def filter(x: Vector) = x.id == >> >> >>> (getIterationRuntimeContext.getSuperstepNumber) >> >> >>> }) >> >> >>> val residual_2 = workset filter {_.id != -1} >> >> >>> val sum = VectorDataSet.add(old_sum, current) >> >> >>> >> >> >>> (sum map (new RichMapFunction[Vector, Vector]{ >> >> >>> def map(x: Vector) = new >> >> >>> Vector(getIterationRuntimeContext.getSuperstepNumber, x.values) >> >> >>> }), >> >> >>> residual_2 union sum) >> >> >>> } >> >> >>> >> >> >>> The error: >> >> >>> >> >> >>> org.apache.flink.compiler.CompilerException: Error: The step >> >> >>> function >> >> >>> does not reference the solution set. >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:868) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.compiler.PactCompiler$GraphCreatingVisitor.postVisit(PactCompiler.java:622) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.api.common.operators.DualInputOperator.accept(DualInputOperator.java:283) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.api.common.operators.SingleInputOperator.accept(SingleInputOperator.java:202) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.api.common.operators.GenericDataSinkBase.accept(GenericDataSinkBase.java:286) >> >> >>> at org.apache.flink.api.common.Plan.accept(Plan.java:281) >> >> >>> at >> >> >>> >> >> >>> org.apache.flink.compiler.PactCompiler.compile(PactCompiler.java:517) >> >> >>> at >> >> >>> >> >> >>> org.apache.flink.compiler.PactCompiler.compile(PactCompiler.java:466) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:196) >> >> >>> at >> >> >>> >> >> >>> >> >> >>> org.apache.flink.client.program.Client.getOptimizedPlan(Client.java:209) >> >> >>> 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) >> >> >>> >> >> >>> Thanks! >> >> >>> Cheers, >> >> >>> Max >> >> >> >> >> >> >> >> > >> > >> > > >
