Cool first version Felix :-)

On Wed, Aug 10, 2016 at 6:48 PM, Stephan Ewen <se...@apache.org> wrote:

> Cool, nice results!
>
> For the iteration unspecialization - we probably should design this hand in
> hand with the streaming fault tolerance, as they share the notion of
> "intermediate result versions".
>
>
> On Wed, Aug 10, 2016 at 6:09 PM, Felix Neutatz <neut...@googlemail.com>
> wrote:
>
> > 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
> > > > > > > > > > >
> > > > > > > > > >
> > > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Reply via email to