Hello all, I've been recently working on a flink api for iterative fixpoint programs and I would like to consult with you, regarding some issues :)
The idea is to provide an easy way for users to write iterative programs of the form x = f(x, D), where x is a set of parameters to be refined, f is the step function and D is the set of dependencies among parameters. >From the user point of view, the api looks a lot like Spargel and the user only needs to provide the initial state of the parameters, the dependencies dataset, an implementation of the step function and the maximum number of iterations. In the background, I spawn a Bulk iteration and while it's running, I approximate the cost of an equivalent Delta iteration plan. As the fixpoint problem gets closer to convergence, this cost diminishes, until it (probably) becomes smaller than the Bulk iteration cost. This is implemented inside a convergence criterion and when it is met, the execution switches to the Delta iteration plan until convergence or until the maximum number of iterations is reached. You can find the current state of the code in my local branch, under stratosphere-fixpoint: https://github.com/vasia/incubator-flink/tree/cost_model One of the problems I have is that when the Bulk iteration cost convergence criterion is met, I cannot find a way to figure out how many iterations have been executed until then. If I use an aggregator, I can only get the result after both the Bulk and Delta iterations have finished (they belong to the same job) and if I use an iteration aggregator, I cannot access its value after the Bulk iteration is finished. Any ideas on possible solutions? The second issue is that the fixpoint api should also support dependencies datasets with or without weights (Tuple2 or Tuple3). Similarly to Spargel, I currently provide 2 separate methods for this and consequently 2 FixpointIteration constructors internally and duplicate methods to support both cases. Any idea on how I could handle this more elegantly? Thanks! Cheers, V.
