You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by VenturaDelMonte <gi...@git.apache.org> on 2016/11/04 15:24:17 UTC
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
GitHub user VenturaDelMonte opened a pull request:
https://github.com/apache/flink/pull/2756
[FLINK-4997] Extending Window Function Metadata
This PR aims to introduce what discussed in [FLIP-2](https://cwiki.apache.org/confluence/display/FLINK/FLIP-2+Extending+Window+Function+Metadata).
WindowedStream apply methods have been overloaded in order to support ProcessWindowFunction (and its rich counterpart).
Streaming runtime internals have been modified in order to support the new function, however fully backward compatibility to WindowFunction (and its rich counterpart) is guaranteed by silently wrapping it with a ProcessWindowFunction.
This PR implementation strictly follow what decided in the FLIP, nothing has been changed for AllWindowedStream.
Windows documentation has been overhauled in order to illustrate ProcessWindowFunction API.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/VenturaDelMonte/flink flip-2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2756.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2756
----
commit 5a33fe2b9e5c23e8529964a489465f51410432df
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-09-21T14:04:00Z
Merge remote-tracking branch 'apache/master'
commit 39a8bb9433edf5bb9adb850c667b71ef8d25c6b6
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-09-22T08:31:58Z
Merge remote-tracking branch 'upstream/master'
commit 2507a71e2a40b05b3e5c7507a1c32d6678e07810
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-09-22T09:40:40Z
Merge remote-tracking branch 'upstream/master'
commit 57d0bca5681e5ea0ba63f3b95fe4f949af3734de
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-10-04T15:03:55Z
Merge remote-tracking branch 'upstream/master'
commit 9f55a1e3e56b48d9dc5a4d1b3109b41e1c89ce5d
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-11-02T09:14:52Z
Merge remote-tracking branch 'upstream/master'
commit 9a71d092ad06c3592355ad11dfb7bd4b982ded9f
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-11-03T20:44:56Z
[FLINK-4997] [streaming] Extending window function metadata introducing ProcessWindowFunction
commit 878983074d364ea0d340bd3497343f286d28e3db
Author: Ventura Del Monte <ve...@gmail.com>
Date: 2016-11-04T14:29:52Z
[FLINK-4997] [docs] improved windows documentation explaining the new ProcessWindowFunction API
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r86973108
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/functions/InternalWindowFunction.java ---
@@ -46,5 +46,5 @@
*
* @throws Exception The function may throw exceptions to fail the program and trigger recovery.
*/
- public abstract void apply(KEY key, W window, IN input, Collector<OUT> out) throws Exception;
+ public abstract void process(KEY key, W window, IN input, Collector<OUT> out) throws Exception;
--- End diff --
I think this PR is good as a first step but I really want to see the benefits of introducing the new interface, especially for users. I don't see how the window metadata is extended with the current implementation. That's my $0.02. @aljoscha is more of an expert here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r88021488
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -800,4 +907,39 @@ public StreamExecutionEnvironment getExecutionEnvironment() {
public TypeInformation<T> getInputType() {
return input.getType();
}
+
+ private static <T, R, K, W extends Window> ProcessWindowFunction<T, R, K, W> wrapWindowFunction(final WindowFunction<T, R, K, W> cleanedFunction) {
--- End diff --
Aljoscha you read my mind, I was just introducing InternalWindowFunction in these other WindowOperators in order to have a flatter implementation flatter. Regarding their removal, I checked that new work and I agree with you.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2756
@VenturaDelMonte Yes to `AllWindowedStream` (Issue, PR).
And yes, we need to extend the interface of `InternalWindowFunction` in the future and we can also change the name to process in a separate commit. I just wanted to reduce the amount of unrelated changes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on the issue:
https://github.com/apache/flink/pull/2756
Hi,
Thank you for both your review and your suggestions! It is good to learn something new everyday!
Regarding InternalWindowFunction, that's alright, although we may need to change it in the future in order to support more metadata.
About AllWindowedStream, of course I would like to, should I open a new issue and make a new PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r88001538
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -800,4 +907,39 @@ public StreamExecutionEnvironment getExecutionEnvironment() {
public TypeInformation<T> getInputType() {
return input.getType();
}
+
+ private static <T, R, K, W extends Window> ProcessWindowFunction<T, R, K, W> wrapWindowFunction(final WindowFunction<T, R, K, W> cleanedFunction) {
--- End diff --
I see, I'm afraid I didn't communicate that very well.
For `WindowOperator` the user facing `WindowFunction` is already abstracted from the actual `WindowOperator` by the `InternalWindowFunction`. I just realised that these "AlignedProcessingTimeWindowOperators" don't have that abstraction and directly take a `WindowFunction`, there we would have to introduce some level of nesting, either by wrapping a `WindowFunction` in a `ProcessWindowFunction` or by also using `InternalWindowFunction` there. Thing is, though, that these operators will be removed because their state cannot easily made repartitionable (see the work that we did on key-groups and repartitionable state).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on the issue:
https://github.com/apache/flink/pull/2756
@aljoscha @VenturaDelMonte This seems to be taking a long time. What is the progress now ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r88021648
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -800,4 +907,39 @@ public StreamExecutionEnvironment getExecutionEnvironment() {
public TypeInformation<T> getInputType() {
return input.getType();
}
+
+ private static <T, R, K, W extends Window> ProcessWindowFunction<T, R, K, W> wrapWindowFunction(final WindowFunction<T, R, K, W> cleanedFunction) {
--- End diff --
\U0001f603
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on the issue:
https://github.com/apache/flink/pull/2756
@aljoscha ok, then let me know, please, if I can help somehow.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2756
The changes look very good! \U0001f44d
I'll rebase on top of the recent changes that introduced new `fold()`/`reduce()` and will also reshuffle/squash the commits a bit to make the history clearer. I hope that's alright.
I'll also change the name of the `process()` method in `InternalWindowFunction` back to `apply()` so that we have less changes and the core of the changes is more obvious. You could change that name in a future PR if you'd like but it's purely internal. I hope this is also alright.
A word on commit titles: we normally use imperative mood, i.e. "Extend WindowFunction Metadata" instead of "Extending ..." or "Extends ...". Initially, this might seem strange but you can think of a commit as being the command to do something. This is a good read on commit messages: http://chris.beams.io/posts/git-commit/ I'm not writing so much to discourage you, I just like good commit messages. And I hope you'll keep up the good work in the code. \U0001f603
Btw, do you also want to do the same changes for `AllWindowedStream`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on the issue:
https://github.com/apache/flink/pull/2756
@VenturaDelMonte Any updates here ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte closed the pull request at:
https://github.com/apache/flink/pull/2756
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on the issue:
https://github.com/apache/flink/pull/2756
@VenturaDelMonte sorry, it was my suggestion to change `InternalWindowFunction#apply` to `process`. Anyway, great work \U0001f44d
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2756
@VenturaDelMonte, that could require yet more tests, yes. We'll see once I'm done with those for the regular windows.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on the issue:
https://github.com/apache/flink/pull/2756
@manuzhang thank you for your feedback and no problem!
@aljoscha \U0001f44d
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by manuzhang <gi...@git.apache.org>.
Github user manuzhang commented on the issue:
https://github.com/apache/flink/pull/2756
R: @aljoscha could you take a look ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r87801992
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -800,4 +907,39 @@ public StreamExecutionEnvironment getExecutionEnvironment() {
public TypeInformation<T> getInputType() {
return input.getType();
}
+
+ private static <T, R, K, W extends Window> ProcessWindowFunction<T, R, K, W> wrapWindowFunction(final WindowFunction<T, R, K, W> cleanedFunction) {
--- End diff --
There is already `WrappingFunction`, that can be used for these kinds of purposes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2756
@manuzhang I finished reviewing and the code is good to go but I'm currently adding more tests before merging this because I realised that we are lacking there a bit.
See https://issues.apache.org/jira/browse/FLINK-5237
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r87802539
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -459,7 +545,28 @@ public WindowedStream(KeyedStream<T, K> input,
* @param resultType Type information for the result type of the window function
* @return The data stream that is the result of applying the window function to the window.
*/
- public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, WindowFunction<R, R, K, W> function, TypeInformation<R> resultType) {
+ public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction,
--- End diff --
I think we can already implement https://issues.apache.org/jira/browse/FLINK-3869 (see my last comment there) for the `ProcessWindowFunction`. I.e. don't have `apply(ReduceFunction, WindowFunction)` but `reduce(ReduceFunction, ProcessWindowFunction)` (same for fold).
Also, `apply(R, FoldFunction, WindowFunction)` has a bug, in that it is too restrictive, see the Jira issue I linked.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r87975777
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -459,7 +545,28 @@ public WindowedStream(KeyedStream<T, K> input,
* @param resultType Type information for the result type of the window function
* @return The data stream that is the result of applying the window function to the window.
*/
- public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, WindowFunction<R, R, K, W> function, TypeInformation<R> resultType) {
+ public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction,
--- End diff --
Thx for the review, first of all. I see your point here. I will introduce these ProcessWindowFunction-specific fold/reduce methods, w/o altering the old WindowFunction API of course.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on the issue:
https://github.com/apache/flink/pull/2756
@manuzhang the other PR is ready since few days, you can find it here: #2946 .
@aljoscha do you plan to add more tests also for AllWindowedStream?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2756: [FLINK-4997] Extending Window Function Metadata
Posted by VenturaDelMonte <gi...@git.apache.org>.
Github user VenturaDelMonte commented on a diff in the pull request:
https://github.com/apache/flink/pull/2756#discussion_r87974146
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---
@@ -800,4 +907,39 @@ public StreamExecutionEnvironment getExecutionEnvironment() {
public TypeInformation<T> getInputType() {
return input.getType();
}
+
+ private static <T, R, K, W extends Window> ProcessWindowFunction<T, R, K, W> wrapWindowFunction(final WindowFunction<T, R, K, W> cleanedFunction) {
--- End diff --
Yes, I know about it, but having a WrappingFunction<WindowFunction> would not be possible here because we need to create a ProcessWindowFunction. Anyway as I would get rid of the triple nesting InternalWindowFunction(ProcessWindowFunction(WindowFunction))), this is no longer an issue. By the way, I added this triple nesting because I thought you wanted to completely decouple the WindowFunction interface from runtime internals.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2756: [FLINK-4997] Extending Window Function Metadata
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2756
I finally merged it! Thanks for staying with this for so long and working on this, @VenturaDelMonte \U0001f603
Hopefully, we can now make faster progress.
Could you please close this PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---