You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by chenqin <gi...@git.apache.org> on 2016/12/10 01:51:11 UTC
[GitHub] flink pull request #2982: [FLINK-4460] Side Outputs in Flink
GitHub user chenqin opened a pull request:
https://github.com/apache/flink/pull/2982
[FLINK-4460] Side Outputs in Flink
[FLIP-13](https://cwiki.apache.org/confluence/display/FLINK/FLIP-13+Side+Outputs+in+Flink) Expose sideOutput with `OutputTag<T>`,
For those userFunction provide `Collector<OUT> collector` as a parameter,
- it offer a util class`CollectorWrapper<OUT> wrapper = new CollectorWrapper<OUT>(collector);` which can write sideOutput element `wrapper.collect(OutputTag<SIDEOUT> tag, sideout)` as well as `getSideOutput(OutputTag<SIDEOUT> tag)` in `singleStreamOutputOpeator` and get sideOutput DataStream<SIDEOUT>.
- each OutputTag with same type can have different value, getSideOutput will only expose element with exact same OutputTag type and value.
sideOutput Late arriving event if
- time characteristic set to eventTime
- all assigned window(s) isLate return(s) true
- event timestamp no later than currentWatermark+ allowedLateness)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/chenqin/flink flip
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2982.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 #2982
----
commit de674f19fcbe9955cb4208ef0938fe5b0f7adc90
Author: Chen Qin <qi...@fgmail.com>
Date: 2016-10-21T19:38:04Z
allow mutpile output stream
commit 3d91e6c69dbfbcb2c73dcc37ac2d8ed637a374eb
Author: Chen Qin <cq...@uber.com>
Date: 2016-11-29T21:24:09Z
Merge branch 'master' into flip
commit 977b2d7fc54e1f9663a5ceb8a62ed2af5a955ca6
Author: Chen Qin <cq...@uber.com>
Date: 2016-12-01T22:19:56Z
allow mutiple OutputTag with same type
implement windowopeator late arriving events
add unit/integration tests
----
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by jgrier <gi...@git.apache.org>.
Github user jgrier commented on the issue:
https://github.com/apache/flink/pull/2982
Nice :)
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
@aljoscha,
diff has been updated and merged to support `apply(Function, lateElementsOutput);`
`Collector` refactor seems not viable option as stated above.
Seems travis-ci timeouted here, is there a way to increase timeout setting?
https://travis-ci.org/chenqin/flink/builds/201329246
Chen
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
Wow, finally in!
It was fun and thanks for you help along every step of the journey!
Chen
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2982
@chenqin Would you like your commits to be attributed to "Chen Qin <qi...@gmail.com>"
or "Chen Qin <cq...@uber.com>"? I see both in your set of commits?
I'm finally putting everything together and hopefully merging soon since the mailing list discussion seems to favour our approach. \U0001f604
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2982
@chenqin
I rebased and consolidated your PR a bit and played around with the APIs.
Some of the changes:
- Separation into internal changes/window operator changes and user-function changes. I have a prototype commit that exposes side outputs using `ProcessFunction`.
- I removed `CollectorWrapper`/`RichCollector` in favour of the `ProcessFunction` approach
- The internal implementation now doesn't store the `OutputTag` on the `StreamRecord` but instead adds an additional method on `Output` that should be used for emitting data to a side output. Side outputs now also work with chaining.
- `WindowedStream` is changed to add a `sideOutputLateData()` method that is used to specify that late data side output is required. This is more general than putting it into the method signature of `apply()` because it will simply work for all different window types.
I quickly talked to @StephanEwen and we agreed that we need to further think about how we want to expose side outputs for user-defined functions. Especially, we have to think about what this addition means for the `split()`/`select()` pattern. I will also do another change where `sideOutputLateData()` is not required but instead will be added when the late-date stream is requested.
What do you think?
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
@aljoscha Thanks for your time. We can chat more after 1.2 release!
I think it makes sense to extends Collector, even though we may not remove collect(T obj) due to API compability issue in 1.X. Per @fhueske comments in [FLIP-13 email thread](http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/Discuss-FLIP-13-Side-Outputs-in-Flink-td14204.html)
Only point I would like to add: there seems decent amount of refactor to replace underlining output chains using collect(tag, element), yet seems reasonable investment moving forward (multiple inputs/ multiple outputs)
`tooLateEvents()` method is something added for user's convenience. should be fine to remove if doesn't gain much benefit. `LateArrivingTag` share same type as input (which is like already fixed once input type defined). Add late arriving tag within apply method seems redudant. In fact, without any changes to this diff, user also be able to access late arriving events via following way.
`
OutputTag<T> lateElementsOutput = new LateArrivingOutputTag<T>();
DataStream<T> input = ...
SingleOutputStreamOperator<O> windowed = input
.keyBy(...)
.window(...)
.apply(Function);
DataStream<T> lateElements = windowed.getSideOutput(lateElementsOutput);
`
Thanks,
Chen
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin closed the pull request at:
https://github.com/apache/flink/pull/2982
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2982
@chenqin I had a quick look at the implementation and it looks quite good. I'll look at it in more detail once the 1.2 release is out and then I'll also have more thorough comments.
These are some quick comments off the top of my head:
- I think we can extend `Collector` with a `collect(OutputTag, T)` method. Then we wouldn't need the extra `RichCollector` and `CollectorWrapper` to work around that.
- For `WindowedStream` I would like to have something like this:
```
OutputTag<T> lateElementsOutput = ...;
DataStream<T> input = ...
SingleOutputStreamOperator<O> windowed = input
.keyBy(...)
.window(...)
.apply(Function, lateElementsOutput);
DataStream<T> lateElements = windowed.getSideOutput(lateElementsOutput);
```
or maube something else if we find a better Idea. With the `WindowedStream.tooLateElements()` this would instantiate an extra `WindowOperator` just for getting late elements while another window operator would be responsible for processing the actual elements. That seems wasteful.
What do you think?
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
@aljoscha
Nice! Let's do "Chen Qin qinnchen@gmail.com"
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
cc @aljoscha
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by chenqin <gi...@git.apache.org>.
Github user chenqin commented on the issue:
https://github.com/apache/flink/pull/2982
@aljoscha Looks good to me \U0001f44d
I briefly looked at your git branch, a minor comment would be adding comments to `sideOutputLateData` so user get better idea when they opt-in to late arriving event stream.
Initial late arriving event is decided by comparing watermark & eventTime, do you think there is a need to allow user pass a kinda `Evaluator` and enable user sideOutput any kind of sideOutputs?
`window.sideOutput(OutputTag<IN>, Evaluator)`
`interface Evaluator{ MergedWindows, key, watermark}`
- Regarding `split` `select`, I think there is a chance of consolidate select and build upon `OutputTag`, but might be out of this PR's scope.
- Regarding to `WindowStream`, I am a bit confused to figure out if I use `allowedlateness` and `sideOutputLateData` at same time.
Thanks,
Chen
---
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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2982
@chenqin I finally merged it. Could you please also close this PR?
And thanks again for working on this for so long! \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 #2982: [FLINK-4460] Side Outputs in Flink
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2982
Thanks for looking at that! I'll open a new discussion thread on the Mailing lists to discuss Side Outputs and split/select and how we're going to proceed with that.
Regarding your other questions: I think we might add such an `Evaluator` interface in the future but for now I would like to keep it simple and see if that works for people. And yes, a user would have to use `allowedLateness` and `sideOutputLateData` at the same time if they want to use late data, or they can go with the default allowed lateness of zero and also get the late data as a side output.
---
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.
---