flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Till Rohrmann <trohrm...@apache.org>
Subject Re: [DISCUSS] FLIP-5 Only send data to each taskmanager once for broadcasts
Date Tue, 09 Aug 2016 14:05:49 GMT
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 <neutatz@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 <trohrmann@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 <neutatz@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 <sewen@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 <
> > neutatz@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@ce9fc38f08a5cc9e93431a9cbf740dcf
> 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 <sewen@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 <
> > > neutatz@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
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message