Re: Flink stops deploying jobs on normal iteration

Posted by Nguyen Xuan Truong on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Flink-stops-deploying-jobs-on-normal-iteration-tp7821p7828.html

Hi Vasia,

Thank you very much for your explanation :). When running with small maxIteration, the job graph that Flink executed was optimal. However, when maxIterations was large, Flink took very long time to generate the job graph. The actually time to execute the jobs was very fast but the time to optimize and schedule the jobs was slow.

Regarding your suggestion, I didn't use iterate/iterateDelta because I need to access the intermediate results within an iteration (the topDistance in my pseudo-code). As you said before, Flink does not support that feature, so I wondered if you have a workround for interate or iterateDelta? 

Thanks,

Truong 

On Tue, Jul 5, 2016 at 8:46 PM, Vasiliki Kalavri <[hidden email]> wrote:

Hi Truong,

I'm afraid what you're experiencing is to be expected. Currently, for loops do not perform well in Flink since there is no support for caching intermediate results yet. This has been a quite often requested feature lately, so maybe it will be added soon :)

Until then, I suggest you try implementing your logic using iterate or iterateDelta.

Cheers,

-Vasia.

On 5 July 2016 at 17:11, Nguyen Xuan Truong <[hidden email]> wrote:

Hi,

I have a Flink program which is similar to Kmeans algorithm. I use normal iteration(for loop) because Flink iteration does not allow to compute the intermediate results(in this case the topDistance) within one iteration. The problem is that my program only runs when maxIteration is small. When the maxIterations is big, Flink jobs inside the forloop are not scheduled, deployed or executed. The program hangs forever without any exception, error or log message.

I ran the program on both local and cluster environments, having the same issue. I tried with smaller inputs (points and seeds), having the same issue. 

Does anybody have an idea about what is the problem? (Maybe the forloop creates many Flink jobs?)

Here is the pseudo-code of my program:

 

DataSet[Point] points = env.readTextFile(inputPoints)

DataSet[Point] seeds = env.readTextFile(inputSeeds)

discardNumber: Int = 100

maxIterations: Int = 20 // maxIteration = 30 will hang the program and no Flink job inside the forloop jobs is deployed)

for(iteration <- 1 to maxIterations) {

      val intermediateSeeds = points

        .map()

        .withBroadcastSet(seeds, "seeds")

     

     //topDistance contains only only double value.

      var topDistance = intermediateSeeds

        .mapPartition()

        .first(discardNumber)

        .groupBy()

        .reduceGroup()

      val newSeeds = intermediateSeeds

        .map()

        .groupBy(0)

        .reduce ().withBroadcastSet(topDistance, "topDistance")

        .map()

      seeds = newSeeds

}

val finalResult = seeds.collect()

Thanks,

Truong