Mailing-List: contact dev-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
MIME-Version: 1.0
From: Xiaowei Jiang <xiaoweij@gmail.com>
Date: Thu, 20 Oct 2016 15:53:30 +0800
Message-ID: <CAJ0NHASevws0+y8LJZ1A-a+ZBX9jvj_P05B8T7GKm1bNkkTXQA@mail.gmail.com>
Subject: Add partitionedKeyBy to DataStream
To: dev@flink.apache.org
Content-Type: multipart/alternative; boundary=001a11410bfcc3886c053f473870
archived-at: Thu, 20 Oct 2016 07:58:42 -0000

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

After we do any interesting operations (e.g. reduce) on KeyedStream, the
result becomes DataStream. In a lot of cases, the output still has the same
or compatible keys with the KeyedStream (logically). But to do further
operations on these keys, we are forced to use keyby again. This works
semantically, but is costly in two aspects. First, it destroys the
possibility of chaining, which is one of the most important optimization
technique. Second, keyby will greatly expand the connected components of
tasks, which has implications in failover optimization.

To address this shortcoming, we propose a new operator partitionedKeyBy.

DataStream {
    public <K> KeyedStream<T, K> partitionedKeyBy(KeySelector<T, K> key)
}

Semantically, DataStream.partitionedKeyBy(key) is equivalent to
DataStream.keyBy(partitionedKey) where partitionedKey is key plus the
taskid as an extra field. This guarantees that records from different tasks
will never produce the same keys.

With this, it's possible to do

ds.keyBy(key1).reduce(func1)
    .partitionedKeyBy(key1).reduce(func2)
    .partitionedKeyBy(key2).reduce(func3);

Most importantly, in certain cases, we will be able to chains these into a
single vertex.

Please share your thoughts. The JIRA is at https://issues.apache.org/j
ira/browse/FLINK-4855

Xiaowei

--001a11410bfcc3886c053f473870--