Mailing-List: contact user-help@flink.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.incubator.apache.org
MIME-Version: 1.0
In-Reply-To: 
 <CAN0XJzM8icgztgx279vs2NApRmqsAZZwSaBN7CVBd8Jy8Ya9-A@mail.gmail.com>
References: 
 <CAN0XJzMg4ABW_99ctbNxXHJa0+gWmLwveXtR=PvGfKhHKG-1Vw@mail.gmail.com>
	<CANMXwW1q37jn6wot5wyit68Ozoac6_upR9=GWRe_xxqe6_vewA@mail.gmail.com>
	<CAN0XJzM8icgztgx279vs2NApRmqsAZZwSaBN7CVBd8Jy8Ya9-A@mail.gmail.com>
Date: Mon, 10 Nov 2014 10:50:28 +0100
Message-ID: 
 <CAAdrtT12xPh+qbdLWJPyr0NqE11T=XE8ZJyvA_c-26nVN9XuJg@mail.gmail.com>
Subject: Re: PartitionByHash and usage of KeySelector
From: Fabian Hueske <fhueske@apache.org>
To: "user@flink.incubator.apache.org" <user@flink.incubator.apache.org>
Content-Type: multipart/alternative; boundary=001a1139d682afb3fc05077e1784

--001a1139d682afb3fc05077e1784
Content-Type: text/plain; charset=UTF-8

Hi Stefano,

I'm not sure if we use the same terminology here. What you call
partitioning might be called grouping in Flinks API / documentation.

Grouping builds groups of element that share the same key. This is a
deterministic operation.
Partitioning distributes elements over a set of machines / parallel
workers. If this is done using hash partitioning, Flink determines the
parallel worker for an element by hashing the element's partition key (
mod(hash(key), #workers) ). Consequently, all elements with the same
partition key will be shipped to the same worker, BUT also all other
elements for which mod(hash(key), #workers) is the same will be shipped to
the same worker. If you partition map over these partitions all of these
elements will be mixed. If the number of workers (or the hash function)
changes, partitions will look different. When grouping all elements of the
group will have the same key (and all elements with that key will be in the
group).

Flink's cross operator builds a dataset wide cross product. It does not
respect groups (or partitions). If you want to build a cross product within
a group, you can do that with a groupReduce which requires to hold all
elements of the group in memory or manually spill them to disk in your UDF.
Alternatively, you can use a self join (join a data set with itself) which
will give you all pairs of the CP in individual function calls. However,
Flink is currently not treating self joins special, such that the
performance could be optimized. You'll also get symmetric pairs (a-b, b-a,
a-a, b-b, for two element a, b with the same join key).

If it is possible to combine the marco-parameter keys and the
minor-blocking keys into a single key, you could specify a key-selector
function x() and either do
- dataSet.groupBy(x).reduceGroup( *read full group into memory, and apply
expensive function to each pair of elements* ); or
- dataSet.join(dataSet).where(x).equalTo(x).join( *check of symmetric pair
and apply expensive compare function* ).

BTW. there was a similar use case a few days back on the mailing list.
Might be worth reading that thread [1].
Since there this is the second time that this issue came up, we might
consider to add better support for group-wise cross operations.

Cheers, Fabian

[1]
http://apache-flink-incubator-mailing-list-archive.1008284.n3.nabble.com/load-balancing-groups-td2287.html

--001a1139d682afb3fc05077e1784
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div>Hi Stefano,</div><div><br></div><div>I&#39;m not sure=
 if we use the same terminology here. What you call partitioning might be c=
alled grouping in Flinks API / documentation.</div><div><br></div><div>Grou=
ping builds groups of element that share the same key. This is a determinis=
tic operation. </div><div>Partitioning distributes elements over a set of m=
achines / parallel workers. If this is done using hash partitioning, Flink =
determines the parallel worker for an element=C2=A0by hashing the element&#=
39;s partition key ( mod(hash(key), #workers) ). Consequently, all elements=
 with the same partition key will be shipped to the same worker, BUT also a=
ll other elements for which mod(hash(key), #workers) is the same will be sh=
ipped to the same worker. If you partition map over these partitions all of=
 these elements will be mixed. If the number of workers (or the hash functi=
on) changes,=C2=A0partitions will look different.=C2=A0When grouping all el=
ements of the group will have the same key (and all elements with that key =
will be in the group).</div><div><br></div><div>Flink&#39;s cross operator =
builds a dataset wide cross product. It does not respect groups (or partiti=
ons). If you want to build a cross product within a group, you can do that =
with a groupReduce which requires to hold all elements of the group in memo=
ry or manually spill them to disk in your UDF. Alternatively, you can use a=
 self join (join a data set with itself) which will give you all pairs of t=
he CP in individual function calls. However, Flink is currently not treatin=
g self joins special, such that the performance could be optimized. You&#39=
;ll also get symmetric pairs (a-b, b-a, a-a, b-b, for two element a, b with=
 the same join key).</div><div><br></div><div>If it is possible to combine =
the marco-parameter keys and the minor-blocking keys into a single key, you=
 could specify a key-selector function x() and either do</div><div>- dataSe=
t.groupBy(x).reduceGroup( *read full group into memory, and apply expensive=
 function to each pair of elements* ); or</div><div>- dataSet.join(dataSet)=
.where(x).equalTo(x).join( *check of symmetric pair and apply expensive com=
pare function* ).</div><div><br></div><div>BTW. there was a similar use cas=
e a few days back on the mailing list. Might be worth reading that thread [=
1].</div><div>Since there this is the second time that this issue came up, =
we might consider to add better support for group-wise cross operations.</d=
iv><div><br></div><div>Cheers, Fabian</div><div><br></div><div>[1] <a href=
=3D"http://apache-flink-incubator-mailing-list-archive.1008284.n3.nabble.co=
m/load-balancing-groups-td2287.html">http://apache-flink-incubator-mailing-=
list-archive.1008284.n3.nabble.com/load-balancing-groups-td2287.html</a></d=
iv><div><br></div><div><br></div></div>

--001a1139d682afb3fc05077e1784--