Hi everybody, I found a quick and dirty way to make the blocking subpartition readable by multiple readers. In the JobGraph generation I make all broadcast partitions blocking (see more details here: https://github.com/FelixNeutatz/incubator-flink/commits/blockingMultipleReads). I want to point out that this branch is only experimental!
This works for the simple Map().withBroadcastSet() use case. To test this approach, I run our peel bundle flink-broadcast ( https://github.com/TU-Berlin-DIMA/flink-broadcast) on the IBM power cluster. Ibm-power has 8 nodes and we scale the number of slots per node from 1 - 16: broadcast.ibm-power-1 broadcast.01 6597.3333333333 broadcast.ibm-power-1 broadcast.02 5997 broadcast.ibm-power-1 broadcast.04 6576.6666666667 broadcast.ibm-power-1 broadcast.08 7024.3333333333 broadcast.ibm-power-1 broadcast.16 6933.3333333333 The last row is the averaged run time in milliseconds over 3 runs. You can clearly see, that the run time stays constant :) As discussed, this approach doesn't work yet for native iterations (see FLINK-1713). So in the next weeks I will work on the native iterations as Stephan proposed. Best regards, Felix 2016-08-09 21:29 GMT+07:00 Stephan Ewen <se...@apache.org>: > I agree with Till. Changing the basic data exchange mechanism would screw > up many other ongoing efforts, like more incremental recovery. > > It seems to make this properly applicable, we need to first un-specialize > the iterations. > > (1) Allow for "versioned" intermediate results, i.e., result-x-superstep1, > result-x-superstep2, result-x-superstep3, result-x-superstep4, ... > We need something similar for fined grained recovery in streaming > (result-x-checkpoint1, result-x-checkpoint2, result-x-checkpoint3, > result-x-checkpoint4, ...) so it may be worth addressing that soon anyways. > > (2) Make iterations not dependent on the special local back channel. > Then we can simply schedule iterations like all other things. > > (3) Do the actual FLIP-5 proposal > > > That's quite an effort, but I fear all else will break the engine and other > efforts. > > Best, > Stephan > > > > > > On Tue, Aug 9, 2016 at 4:05 PM, Till Rohrmann <trohrm...@apache.org> > wrote: > > > Hi Felix, > > > > if we cannot work around the problem with blocking intermediate results > in > > iterations, then we have to make FLINK-1713 a blocker for this new issue. > > But maybe you can also keep the current broadcasting mechanism to be used > > within iterations only. Then we can address the iteration problem later. > > > > Cheers, > > Till > > > > On Tue, Aug 9, 2016 at 3:54 PM, Felix Neutatz <neut...@googlemail.com> > > wrote: > > > > > Hi Till, > > > > > > thanks for the fast answer. I also think this should be the way to go. > So > > > should I open a new jira "Make blocking SpillableSubpartition able to > be > > > read multiple times". Moreover should I mark this jira and FLINK-1713 > > > <https://issues.apache.org/jira/browse/FLINK-1713> as blocking for the > > > broadcast jira? What do you think? > > > > > > Best regards, > > > Felix > > > > > > 2016-08-09 17:41 GMT+07:00 Till Rohrmann <trohrm...@apache.org>: > > > > > > > Hi Felix, > > > > > > > > I'm not sure whether PipelinedSubpartition should be readable more > than > > > > once because then it would effectively mean that we materialize the > > > > elements of the pipelined subpartition for stragglers. Therefore, I > > think > > > > that we should make blocking intermediate results readable more than > > > once. > > > > This will also be beneficial for interactive programs where we > continue > > > > from the results of previous Flink jobs. > > > > > > > > It might also be interesting to have a blocking mode which schedules > > its > > > > consumers once the first result is there. Thus, having a mixture of > > > > pipelined and blocking mode. > > > > > > > > Cheers, > > > > Till > > > > > > > > On Tue, Aug 9, 2016 at 4:40 AM, Felix Neutatz < > neut...@googlemail.com> > > > > wrote: > > > > > > > > > Hi Stephan, > > > > > > > > > > I did some research about blocking intermediate results. It turns > out > > > > that > > > > > neither PipelinedSubpartition (see line 178) nor blocking > > intermediate > > > > > results (see SpillableSubpartition line: 189) can be read multiple > > > times. > > > > > Moreover blocking intermediate results are currently not supported > in > > > > > native iterations (see https://issues.apache.org/ > > > jira/browse/FLINK-1713 > > > > ). > > > > > So there are three ways to solve this: > > > > > 1) We extend Pipelined subpartitions to make it possible to read > them > > > > > multiple times > > > > > 2) We extend Blocking subpartitions to make it possible to read > them > > > > > multiple times, but then we also have to fix FLINK-1713. So we can > > use > > > > > broadcasts in native iterations > > > > > 3) We create one pipelined subpartition for every taskmanager. > > Problem: > > > > The > > > > > more taskmanager there are, the more redundant data we store, but > the > > > > > network traffic stays optimal. > > > > > > > > > > Thank you for your help, > > > > > Felix > > > > > > > > > > 2016-08-01 22:51 GMT+07:00 Stephan Ewen <se...@apache.org>: > > > > > > > > > > > Hi Felix! > > > > > > > > > > > > Hope this helps_ > > > > > > > > > > > > Concerning (1.1) - The producer does not think in term of number > of > > > > > target > > > > > > TaskManagers. That number can, after all, change in the presence > > of a > > > > > > failure and recovery. The producer should, for its own result, > not > > > care > > > > > how > > > > > > many consumers it will have (Tasks), but produce it only once. > > > > > > > > > > > > Concerning (1.2) - Only "blocking" intermediate results can be > > > > consumed > > > > > > multiple times. Data sent to broadcast variables must thus be > > always > > > a > > > > > > blocking intermediate result. > > > > > > > > > > > > Greetings, > > > > > > Stephan > > > > > > > > > > > > > > > > > > On Wed, Jul 27, 2016 at 11:33 AM, Felix Neutatz < > > > > neut...@googlemail.com> > > > > > > wrote: > > > > > > > > > > > > > Hi Stephan, > > > > > > > > > > > > > > thanks for the great ideas. First I have some questions: > > > > > > > > > > > > > > 1.1) Does every task generate an intermediate result partition > > for > > > > > every > > > > > > > target task or is that already implemented in a way so that > there > > > are > > > > > > only > > > > > > > as many intermediate result partitions per task manager as > target > > > > > tasks? > > > > > > > (Example: There are 2 task managers with 2 tasks each. Do we > get > > 4 > > > > > > > intermediate result partitions per task manager or do we get > 8?) > > > > > > > 1.2) How can I consume an intermediate result partition > multiple > > > > times? > > > > > > > When I tried that I got the following exception: > > > > > > > Caused by: java.lang.IllegalStateException: Subpartition 0 of > > > > > > > dbe284e3b37c1df1b993a3f0a6020ea6@ > ce9fc38f08a5cc9e93431a9cbf740d > > cf > > > is > > > > > > being > > > > > > > or already has been consumed, but pipelined subpartitions can > > only > > > be > > > > > > > consumed once. > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.partition. > > > PipelinedSubpartition. > > > > > > createReadView(PipelinedSubpartition.java:179) > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.partition. > > > PipelinedSubpartition. > > > > > > createReadView(PipelinedSubpartition.java:36) > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.partition.ResultPartition. > > > > > > createSubpartitionView(ResultPartition.java:348) > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.partition. > > > > ResultPartitionManager. > > > > > > createSubpartitionView(ResultPartitionManager.java:81) > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.netty. > > > > > PartitionRequestServerHandler. > > > > > > channelRead0(PartitionRequestServerHandler.java:98) > > > > > > > at > > > > > > > > > > > > > > org.apache.flink.runtime.io.network.netty. > > > > > PartitionRequestServerHandler. > > > > > > channelRead0(PartitionRequestServerHandler.java:41) > > > > > > > at > > > > > > > > > > > > > > io.netty.channel.SimpleChannelInboundHandler.channelRead( > > > > > > SimpleChannelInboundHandler.java:105) > > > > > > > > > > > > > > My status update: Since Friday I am implementing your idea > > > described > > > > in > > > > > > > (2). Locally this approach already works (for less than 170 > > > > > iterations). > > > > > > I > > > > > > > will investigate further to solve that issue. > > > > > > > > > > > > > > But I am still not sure how to implement (1). Maybe we > introduce > > a > > > > > > similar > > > > > > > construct like the BroadcastVariableManager to share the > > > RecordWriter > > > > > > among > > > > > > > all tasks of a taskmanager. I am interested in your thoughts :) > > > > > > > > > > > > > > Best regards, > > > > > > > Felix > > > > > > > > > > > > > > 2016-07-22 17:25 GMT+02:00 Stephan Ewen <se...@apache.org>: > > > > > > > > > > > > > > > Hi Felix! > > > > > > > > > > > > > > > > Interesting suggestion. Here are some thoughts on the design. > > > > > > > > > > > > > > > > The two core changes needed to send data once to the > > TaskManagers > > > > > are: > > > > > > > > > > > > > > > > (1) Every sender needs to produce its stuff once (rather > than > > > for > > > > > > every > > > > > > > > target task), there should not be redundancy there. > > > > > > > > (2) Every TaskManager should request the data once, other > > tasks > > > > in > > > > > > the > > > > > > > > same TaskManager pick it up from there. > > > > > > > > > > > > > > > > > > > > > > > > The current receiver-initialted pull model is actually a good > > > > > > abstraction > > > > > > > > for that, I think. > > > > > > > > > > > > > > > > Lets look at (1): > > > > > > > > > > > > > > > > - Currently, the TaskManagers have a separate intermediate > > > result > > > > > > > > partition for each target slot. They should rather have one > > > > > > intermediate > > > > > > > > result partition (saves also repeated serialization) that is > > > > consumed > > > > > > > > multiple times. > > > > > > > > > > > > > > > > - Since the results that are to be broadcasted are always > > > > > "blocking", > > > > > > > > they can be consumed (pulled) multiples times. > > > > > > > > > > > > > > > > Lets look at (2): > > > > > > > > > > > > > > > > - The current BroadcastVariableManager has the > functionality > > to > > > > let > > > > > > the > > > > > > > > first accessor of the BC-variable materialize the result. > > > > > > > > > > > > > > > > - It could be changed such that only the first accessor > > > creates a > > > > > > > > RecordReader, so the others do not even request the stream. > > That > > > > way, > > > > > > the > > > > > > > > TaskManager should pull only one stream from each producing > > task, > > > > > which > > > > > > > > means the data is transferred once. > > > > > > > > > > > > > > > > > > > > > > > > That would also work perfectly with the current failure / > > > recovery > > > > > > model. > > > > > > > > > > > > > > > > What do you think? > > > > > > > > > > > > > > > > Stephan > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Jul 22, 2016 at 2:59 PM, Felix Neutatz < > > > > > neut...@googlemail.com > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > Hi everybody, > > > > > > > > > > > > > > > > > > I want to improve the performance of broadcasts in Flink. > > > > Therefore > > > > > > > Till > > > > > > > > > told me to start a FLIP on this topic to discuss how to go > > > > forward > > > > > to > > > > > > > > solve > > > > > > > > > the current issues for broadcasts. > > > > > > > > > > > > > > > > > > The problem in a nutshell: Instead of sending data to each > > > > > > taskmanager > > > > > > > > only > > > > > > > > > once, at the moment the data is sent to each task. This > means > > > if > > > > > > there > > > > > > > > are > > > > > > > > > 3 slots on each taskmanager we will send the data 3 times > > > instead > > > > > of > > > > > > > > once. > > > > > > > > > > > > > > > > > > There are multiple ways to tackle this problem and I > started > > to > > > > do > > > > > > some > > > > > > > > > research and investigate. You can follow my thought process > > > here: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP- > > > > > > 5%3A+Only+send+data+to+each+taskmanager+once+for+broadcasts > > > > > > > > > > > > > > > > > > This is my first FLIP. So please correct me, if I did > > something > > > > > > wrong. > > > > > > > > > > > > > > > > > > I am interested in your thoughts about how to solve this > > issue. > > > > Do > > > > > > you > > > > > > > > > think my approach is heading into the right direction or > > should > > > > we > > > > > > > > follow a > > > > > > > > > totally different one. > > > > > > > > > > > > > > > > > > I am happy about any comment :) > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > Felix > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
