Return-Path: X-Original-To: apmail-flink-user-archive@minotaur.apache.org Delivered-To: apmail-flink-user-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id D1E2C18CF5 for ; Thu, 13 Aug 2015 08:03:38 +0000 (UTC) Received: (qmail 36014 invoked by uid 500); 13 Aug 2015 08:03:33 -0000 Delivered-To: apmail-flink-user-archive@flink.apache.org Received: (qmail 35934 invoked by uid 500); 13 Aug 2015 08:03:33 -0000 Mailing-List: contact user-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@flink.apache.org Delivered-To: mailing list user@flink.apache.org Received: (qmail 35923 invoked by uid 99); 13 Aug 2015 08:03:33 -0000 Received: from mail-relay.apache.org (HELO mail-relay.apache.org) (140.211.11.15) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 13 Aug 2015 08:03:33 +0000 Received: from mail-wi0-f181.google.com (mail-wi0-f181.google.com [209.85.212.181]) by mail-relay.apache.org (ASF Mail Server at mail-relay.apache.org) with ESMTPSA id D43361A0325 for ; Thu, 13 Aug 2015 08:03:32 +0000 (UTC) Received: by wicja10 with SMTP id ja10so58804318wic.1 for ; Thu, 13 Aug 2015 01:03:30 -0700 (PDT) X-Received: by 10.194.89.5 with SMTP id bk5mr78921301wjb.144.1439453010675; Thu, 13 Aug 2015 01:03:30 -0700 (PDT) MIME-Version: 1.0 References: <1439329285784-2373.post@n4.nabble.com> <55CB247B.9080706@students.uni-marburg.de> In-Reply-To: <55CB247B.9080706@students.uni-marburg.de> From: Aljoscha Krettek Date: Thu, 13 Aug 2015 08:03:21 +0000 Message-ID: Subject: Re: Forward Partitioning & same Parallelism: 1:1 communication? To: user@flink.apache.org Content-Type: multipart/alternative; boundary=089e010d84005f4df5051d2cc541 --089e010d84005f4df5051d2cc541 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: quoted-printable Hi, I looked into it. Right now, when the specified partitioner is "FORWARD" the JobGraph that is generated from the StreamGraph will have the POINT-TO-POINT pattern specified. This doesn't work, however, if the parallelism differs so the operators will not have a POINT-TO-POINT connection in the end. This results in the "REBALANCE" behavior your observed. My PR makes it explicit which connection pattern will be used. It will also properly set the connection to "REBALANCE" if not partitioning is specified and the parallelism of the operators is different. I hope this helps somehow, let us know if you have any other questions. Cheers, Aljoscha On Wed, 12 Aug 2015 at 11:49 Anneke Walter wrote: > Hey M=C3=A1rton, hey Ufuk, > > thank you for your replies, that was very helpful! > > I now have an additional question based on M=C3=A1rton's answer to Ufuk's > question (by the way, I'm currently working only with the streaming API, = so > I am most interested in answers concerning streaming than batch > processing... :-) ) > > In the second link M=C3=A1rton provided [1] it says: > > "This was not very transparent: When you went from low parallelism to hig= h > dop some downstream operators would never get any input." > > A question below the pull request then asks "If a non parallel source is > used does the user need to call rebalance to use all parallel instances > of the downstream operator?" and I don't think that question was explicit= ly > answered. The closest thing to an explicit answer is "[so far] forward wa= s > assumed. This was valid for a change of parallelism, which led to either > the degenerative case of only one downstream instance receiving elements = (1 > to n parallelism)". > > To me, that sounds as if up until right now, in a situation where operato= r > A has lower parallelism than the following downstream operator B (for > example, source A with parallelism 1 and filter B with parallelism 4), no= t > all instances of B would receive output from A if forward partitioning is > used. > > Now, in the docs [2] it says: > * "Forward (default)*: Forward partitioning directs the output data to > the next operator on the same machine (if possible) avoiding expensive > network I/O. _If there are more processing nodes than inputs or vice vers= a > the load is distributed among the extra nodes in a round-robin fashion_. > This is the default partitioner." > > So far, I would've thought that the middle sentence describes that when > forward partitioning is used when the parallelism differs, outputs will b= e > forwarded to the next operator on the same machine where possible, but al= so > distributing some outputs to the extra nodes with round-robin. However, > I've tested the setup describes above (see below) and it seems that Flink > uses "normal" round-robin partitioning (rebalance partitioning) when the > parallelism differs - using round-robin for _all_ outputs, not doing any > "forwarding" (in the forward partitioning sense). Is that correct? > > My little test: 1 Sink, 4 Filters > I tried that with Flink 0.9 and, even though I did not explicitly specify > any partitioning (so the default, forward, should have been used), Flink > apparently uses rebalance partitioning in this case - from the log: > DEBUG StreamingJobGraphGenerator:235 Thread-1 - Parallelism set: 4 for 2 > DEBUG StreamingJobGraphGenerator:235 Thread-1 - Parallelism set: 1 for 1 > DEBUG StreamingJobGraphGenerator:312 Thread-1 - CONNECTED: > RebalancePartitioner - 1 -> 2 > > Same thing happened reversely when I went from 4 filters (B) to 1 sink C, > rebalancing was apparently used. > > So that one problem (concerning downstream operators not receiving output= s > when forward partitioning is used) described in the pull request is > apparently already fixed in 0.9 - or does it only work correctly for the > source/sink connection and not between other operators (I did not have ti= me > to try more scenarios)? > > Again, I would be very happy about some input about if I grasped Flink's > behavior correctly! :-) Thanks in advance! > Nica > > [1] https://github.com/apache/flink/pull/988 > [2] https://ci.apache.org/projects/flink/flink-docs-release-0.9/apis/stre= aming_guide.html#partitioning > > > Am 12.08.2015 um 11:50 schrieb Ufuk Celebi: > > Thanks :) Regarding your answer to Nica: I didn't mean to say that it was > too generic or anything... it was very nice. I was just curious, that's w= hy > I asked. > > On Wed, Aug 12, 2015 at 11:45 AM, M=C3=A1rton Balassi > wrote: > > > Hey Ufuk, > > The shipping strategy name forward is shared between batch and streaming > and Nica did not specify either API, so I tried to give a generic answer. > > I assume that your question is specifically for streaming, in that case: > Yes, streaming is using the pointwise distribution pattern. [1] > Unfortunately your concern is true, currently streaming would leave extra > downstream operator instances idle, but Aljoscha has an open pull request > fixing this issue amongst others. See the discussion here. [2] > > [1]https://github.com/apache/flink/blob/master/flink-staging/flink-stream= ing/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph= /StreamingJobGraphGenerator.java#L320 > [2] https://github.com/apache/flink/pull/988 > > Cheers, > > Marton > > On Wed, Aug 12, 2015 at 11:33 AM, Ufuk Celebi > wrote: > > > Hey Marton, > > out of curiosity: is this using Flink=E2=80=99s =E2=80=9Cpoint=E2=80=9D c= onnections underneath or > is there some custom logic for streaming jobs? > > What happens if operator B has 2 times the parallelism of operator A? For > example if there were parallel tasks A1 and A2 and B1-B4: would A1 send t= o > B1 *and* B2 or just B1? > > =E2=80=93 Ufuk > > On 12 Aug 2015, at 10:39, M=C3=A1rton Balassi = > wrote: > > Dear Nica, > > Yes, forward partitioning means that if subsequent operators share > parallelism then the output of an upstream operator is sent to exactly > one downstream operator. This makes sense for operators working on > individual records, e.g. a typical map-filter pair, because as a > consequence Flink may be able to collocate these operator pairs on the sa= me > physical machine. > > Best, > > Marton > > On Tue, Aug 11, 2015 at 11:41 PM, Nicaz > wrote: > > Hello, > > I have a question about forward partitioning in Flink. > > If Operator A and Operator B have the same parallelism set and forward > partitioning is used for events coming from instances of A and going to > instances of B: > > Will each instance of A send events to _exactly one_ instance of B? > > That is, will all events coming from a specific instance of A go to the > _same_ specific instance of B, and will _all_ instances of B be used? > Or are there any situations where an instance of A will distribute events > to > several different instances of B, or where two instances of A will send > events to the same instance of B (possibly leaving some other instance of > B > unused)? > > I'd be very happy if someone were able to shed some light on this issue. > :-) > > Thanks in advance > Nica > > > > -- > View this message in context:http://apache-flink-user-mailing-list-archiv= e.2336050.n4.nabble.com/Forward-Partitioning-same-Parallelism-1-1-communica= tion-tp2373.html > Sent from the Apache Flink User Mailing List archive. mailing list > archive at Nabble.com. > > > > > > --089e010d84005f4df5051d2cc541 Content-Type: text/html; charset=UTF-8 Content-Transfer-Encoding: quoted-printable
Hi,
I looked into it. Right now, when the specified pa= rtitioner is "FORWARD" the JobGraph that is generated from the St= reamGraph will have the POINT-TO-POINT pattern specified. This doesn't = work, however, if the parallelism differs so the operators will not have a = POINT-TO-POINT connection in the end. This results in the "REBALANCE&q= uot; behavior your observed.

My PR makes it explic= it which connection pattern will be used. It will also properly set the con= nection to "REBALANCE" if not partitioning is specified and the p= arallelism of the operators is different.

I hope t= his helps somehow, let us know if you have any other questions.=C2=A0
=

Cheers,
Aljoscha

On Wed, 12 Aug 2015 at 11:49 Anneke Walter &= lt;Walteran@students.un= i-marburg.de> wrote:
=20 =20 =20
Hey M=C3=A1rton, hey Ufuk,

thank you for your replies, that was very helpful!

I now have an additional question based on M=C3=A1rton's answer to = Ufuk's question (by the way, I'm currently working only with the streaming API, so I am most interested in answers concerning streaming than batch processing... :-) )

In the second link M=C3=A1rton provided [1] it says:

"This was not very transparent: When you went from low parallelism to high dop some downstream operators would never get any input."

A question below the pull request then asks "If a non parallel source is used does the user need to call rebalance to use all parallel instances of the downstream operator?" and I don&= #39;t think that question was explicitly answered. The closest thing to an explicit answer is "[so far] forward was assumed. This was valid f= or a change of parallelism, which led to either the degenerative case of only one downstream instance receiving elements (1 to n parallelism)".

To me, that sounds as if up until right now, in a situation where operator A has lower parallelism than the following downstream operator B (for example, source A with parallelism 1 and filter B with parallelism 4), not all instances of B would receive output from A if forward partitioning is used.

Now, in the docs [2] it says:
"Forward (default): Forward partitioning directs the output data to the next operator on the same machine (if possible) avoiding expensive network I/O. _If there are more processing nodes than inputs or vice versa the load is distributed among the extra nodes in a round-robin fashion_. This is the default partitioner."

So far, I would've thought that the middle sentence describes that when forward partitioning is used when the parallelism differs, outputs will be forwarded to the next operator on the same machine where possible, but also distributing some outputs to the extra nodes with round-robin. However, I've tested the setup describes above (see below) and it seems that Flink uses "normal" round= -robin partitioning (rebalance partitioning) when the parallelism differs - using round-robin for _all_ outputs, not doing any "forwarding&quo= t; (in the forward partitioning sense). Is that correct?

My little test: 1 Sink, 4 Filters
I tried that with Flink 0.9 and, even though I did not explicitly specify any partitioning (so the default, forward, should have been used), Flink apparently uses rebalance partitioning in this case - from the log:
DEBUG StreamingJobGraphGenerator:235 Thread-1 - Parallelism set: 4 for 2
DEBUG StreamingJobGraphGenerator:235 Thread-1 - Parallelism set: 1 for 1
DEBUG StreamingJobGraphGenerator:312 Thread-1 - CONNECTED: RebalancePartitioner - 1 -> 2

Same thing happened reversely when I went from 4 filters (B) to 1 sink C, rebalancing was apparently used.

So that one problem (concerning downstream operators not receiving outputs when forward partitioning is used) described in the pull request is apparently already fixed in 0.9 - or does it only work correctly for the source/sink connection and not between other operators (I did not have time to try more scenarios)?

Again, I would be very happy about some input about if I grasped Flink's behavior correctly! :-) Thanks in advance!
Nica

[1] https://github.com/apache/flink/pull/988
[2] https://ci.apache=
.org/projects/flink/flink-docs-release-0.9/apis/streaming_guide.html#partit=
ioning

Am 12.08.2015 um 11:50 schrieb Ufuk Celebi:
Thanks :) Regarding your answer to Nica: I didn't mean to sa=
y that it was
too generic or anything... it was very nice. I was just curious, that's=
 why
I asked.

On Wed, Aug 12, 2015 at 11:45 AM, M=C3=A1rton Balassi <balassi.marton@gmail.com>
wrote:


Hey Ufuk,

The shipping strategy name forward is shared between batch and streaming
and Nica did not specify either API, so I tried to give a generic answer.

I assume that your question is specifically for streaming, in that case:
Yes, streaming is using the pointwise distribution pattern. [1]
Unfortunately your concern is true, currently streaming would leave extra
downstream operator instances idle, but Aljoscha has an open pull request
fixing this issue amongst others. See the discussion here. [2]

[1]
https://gith=
ub.com/apache/flink/blob/master/flink-staging/flink-streaming/flink-streami=
ng-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGrap=
hGenerator.java#L320
[2] =
https://github.com/apache/flink/pull/988

Cheers,

Marton

On Wed, Aug 12, 2015 at 11:33 AM, Ufuk Celebi <ufuk@data-artisans.com>
wrote:


Hey Marton,

out of curiosity: is this using Flink=E2=80=99s =E2=80=9Cpoint=E2=80=9D con=
nections underneath or
is there some custom logic for streaming jobs?

What happens if operator B has 2 times the parallelism of operator A? For
example if there were parallel tasks A1 and A2 and B1-B4: would A1 send to
B1 *and* B2 or just B1?

=E2=80=93 Ufuk

On 12 Aug 2015, at 10:39, M=C3=A1rton Balassi <balassi.marton@gmail.com>
wrote:

Dear Nica,

Yes, forward partitioning means that if subsequent operators share
parallelism then the output of an upstream operator is sent to exactly
one downstream operator. This makes sense for operators working on
individual records, e.g. a typical map-filter pair, because as a
consequence Flink may be able to collocate these operator pairs on the same
physical machine.

Best,

Marton

On Tue, Aug 11, 2015 at 11:41 PM, Nicaz <Walteran@students.uni-marburg.de

wrote:

Hello,

I have a question about forward partitioning in Flink.

If Operator A and Operator B have the same parallelism set and forward
partitioning is used for events coming from instances of A and going to
instances of B:

Will each instance of A send events to _exactly one_ instance of B?

That is, will all events coming from a specific instance of A go to the
_same_ specific instance of B, and will _all_ instances of B be used?
Or are there any situations where an instance of A will distribute events
to
several different instances of B, or where two instances of A will send
events to the same instance of B (possibly leaving some other instance of
B
unused)?

I'd be very happy if someone were able to shed some light on this issue=
.
:-)

Thanks in advance
Nica



--
View this message in context:
http://apache-flink-user-mailing-list-archive.2336050.n4.na=
bble.com/Forward-Partitioning-same-Parallelism-1-1-communication-tp2373.htm=
l
Sent from the Apache Flink User Mailing List archive. mailing list
archive at Nabble.com.





      

      

    

    

  


--089e010d84005f4df5051d2cc541--