Hi Robert!

Ok, good to know. Thanks!

Cheers!
Max

On Thu, Jan 8, 2015 at 4:16 PM, Robert Metzger wrote:
Hi Max,

No. I think there is nobody in the Flink community who has plans to implement nested iterations in the near future.

On Wed, Jan 7, 2015 at 10:58 AM, Maximilian Alber wrote:
Thanks!

I made a workaround using a pseudo join with the workset. But now I'm back to the nested iteration issue. Is there any chance that this feature will be available in the next time(2-3 weeks)?

Cheers,
Max

On Wed, Jan 7, 2015 at 10:11 AM, Aljoscha Krettek wrote:
Hi,
the problem is that your operations do not depend on the iteration-step-dataset. Your code could be rewritten like this to make it more obvious:

val emptyDataSet = env.fromCollection[Vector](Seq())
// here we call the function
val center = calcCenter(env, X, residual, randoms, -1)

val centerX = (X subtV center) map {_ square}
val x = calcWidthHeight(env, centerX, residual, widthCandidates, center)
val width = x._1
val height = x._2

residual = residual - (getKernelVector(X, center, width) multV height)

val centerOut = center map {x => new Vector(0, x.values)}
val widthOut = width map {x => new Vector(1, x.values)}
val heightOut = height map {x => new Vector(2, x.values)}
val stepModel = centerOut union widthOut union height

// here the loop begins
val model = emptyDataSet.iterate(config.iterations){
stepSet =>

stepSet union (stepModel.map(new RichMapFunction[Vector, Vector]{
def map(x: Vector) = new Vector((getIterationRuntimeContext.getSuperstepNumber-1)*3+x.id, x.values)
}))
}

model map { _ toString } writeAsText config.outFile
}

This means that for the system these operations are considered to be outside the loop, thus you don't have access to the IterationContext.

Regards,
Aljoscha

On Tue, Jan 6, 2015 at 4:13 PM, Maximilian Alber wrote:

ran into this error

java.lang.IllegalStateException: This stub is not part of an iteration step function.

below is my code, the concerning parts are marked. Is it a problem, that the stub is in a function that is called from the iteration step function?

Code:

......

val emptyDataSet = env.fromCollection[Vector](Seq())
// here the loop begins
val model = emptyDataSet.iterate(config.iterations){
stepSet =>
// here we call the function
val center = calcCenter(env, X, residual, randoms, -1)

val centerX = (X subtV center) map {_ square}
val x = calcWidthHeight(env, centerX, residual, widthCandidates, center)
val width = x._1
val height = x._2

residual = residual - (getKernelVector(X, center, width) multV height)

val centerOut = center map {x => new Vector(0, x.values)}
val widthOut = width map {x => new Vector(1, x.values)}
val heightOut = height map {x => new Vector(2, x.values)}
val stepModel = centerOut union widthOut union height

stepSet union (stepModel.map(new RichMapFunction[Vector, Vector]{
def map(x: Vector) = new Vector((getIterationRuntimeContext.getSuperstepNumber-1)*3+x.id, x.values)
}))
}

model 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 randomValue = if(iteration >= 0)
(randoms filter {_.id == iteration})
else
// and this filter function causes the error
(randoms.filter(new RichFilterFunction[Vector]{
def filter(x: Vector) = x.id == (getIterationRuntimeContext.getSuperstepNumber-1)
}))
val ys = ((residual_2 sumV() neutralize) * (randomValue neutralize))

.....

The full errror:

Error: The program execution failed: java.lang.IllegalStateException: This stub is not part of an iteration step function.
at bumpboost.BumpBoost\$\$anon\$2.filter(BumpBoost.scala:119)
at bumpboost.BumpBoost\$\$anon\$2.filter(BumpBoost.scala:118)