[BEAM-7972] Always use Global window in reshuffle and then apply wind… by angoenka · Pull Request #9334 · apache/beam

angoenka · 2019-08-13T23:00:42Z

…ow again.

Please add a meaningful description for your change here

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza	Spark
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

angoenka · 2019-08-13T23:01:41Z

R: @robertwb @y1chi @tweise

angoenka · 2019-08-14T18:22:04Z

R: @lukecwik

y1chi · 2019-08-19T17:50:39Z

LGTM.
thanks for looking into this.

angoenka · 2019-08-20T17:56:00Z

Gentle reminder for the review.
@robertwb @lukecwik

robertwb

This is just needed to work around a Dataflow JRH bug, right?

robertwb · 2019-08-20T23:11:25Z

sdks/python/apache_beam/transforms/util.py

        key, value = element
-        return key, TimestampedValue(value, timestamp)
+        # Transport the window as part of the value and restore it later.
+        return key, TimestampedValue((value, window), timestamp)


Any reason not to use a WindowedValue here?

Not really. I will make it windowed values.

Actually while making the change i realized that the timestamp will be duplicated when using the windowed values without any benefit. Hence dropping the use of windowed value.

Why would the timestamp be duplicated? The (window, timestamp, value) tuple seems best represented by a WindowedValue. (I would be OK with a plain old 3-tuple as well, but only going half way seems odd).

The defaulting to global window should be deleted since the Python SDK now does send a proper windowing strategy (same as Go SDK). The code was added as a migration path to allow for differences in where the Python/Go/Java SDKs were when submitting jobs to Dataflow.

So we should update the reshuffle code to not pass the non standard window from python.

We shouldn't have to, but if the alternative is significant JRH refactoring, then this code should be OK and we can add a comment that we're working around bugs in the Dataflow JRH.

From a quick look, JRH refactoring seems significant.
I would like to keep it simple for now and will continue with changes in Reshuffle transform with additional comment.

Why would the timestamp be duplicated? The (window, timestamp, value) tuple seems best represented by a WindowedValue. (I would be OK with a plain old 3-tuple as well, but only going half way seems odd).

Ohh, I see what you mean. Updated the code.

angoenka

I think this will be needed irrespective of the JRH bug as we don't want to introduce a new windowing function which should be interpreted by the runner.
Here is the code which defaults to globalWindow when its not able to deserialize the windowing strategy. https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowParDoFnFactory.java#L106

angoenka · 2019-08-20T23:43:45Z

sdks/python/apache_beam/transforms/util.py

        key, value = element
-        return key, TimestampedValue(value, timestamp)
+        # Transport the window as part of the value and restore it later.
+        return key, TimestampedValue((value, window), timestamp)


Not really. I will make it windowed values.

angoenka · 2019-08-22T19:37:49Z

Gentle reminder for the review.

robertwb · 2019-08-23T22:44:33Z

OK, yuck, that JRH code is really bad (and possibly buggy). The windowing function need not be interpreted by the runner, it should just note it's non-merging and pass things through.

angoenka · 2019-08-24T06:47:26Z

That piece of JRH seems to be very intricate and I would don't think it has an easy fix without a lot of refactoring.
Do you have any better solution in mind?

lukecwik · 2019-08-24T16:44:56Z

That piece of JRH seems to be very intricate and I would don't think it has an easy fix without a lot of refactoring.
Do you have any better solution in mind?

The defaulting to global window should be deleted since the Python SDK now does send a proper windowing strategy (same as Go SDK). The code was added as a migration path to allow for differences in where the Python/Go/Java SDKs were when submitting jobs to Dataflow.

angoenka · 2019-08-26T21:05:20Z

The defaulting to global window should be deleted since the Python SDK now does send a proper windowing strategy (same as Go SDK). The code was added as a migration path to allow for differences in where the Python/Go/Java SDKs were when submitting jobs to Dataflow.

So we should update the reshuffle code to not pass the non standard window from python.

angoenka · 2019-08-30T05:16:26Z

Request for another pass for the review.

robertwb · 2019-08-30T16:28:39Z

sdks/python/apache_beam/transforms/util.py

        return [
            windowed_value.WindowedValue(
-                (key, value.value), value.timestamp, [window])
+                (key, value.value), value.timestamp, value.windows)


You could do

key, windowed_values = element return [wv.with_value((key, wv.value)) for wv in windowed_values]

robertwb · 2019-08-30T16:28:55Z

sdks/python/apache_beam/transforms/util.py

    ungrouped = pcoll | Map(reify_timestamps)
+
+    # TODO(BEAM-8104) Using global window as one of the standard window.
+    # This is to mitigate the Java Runner Harness limitation to


s/Java Runner Harness/Dataflow Java Runner Harness/

…ow again.

robertwb · 2019-08-30T22:58:38Z

LGTM

…

On Fri, Aug 30, 2019, 3:37 PM Ankur ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In sdks/python/apache_beam/transforms/util.py <#9334 (comment)>: > for value in values] ungrouped = pcoll | Map(reify_timestamps) + + # TODO(BEAM-8104) Using global window as one of the standard window. + # This is to mitigate the Java Runner Harness limitation to done — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#9334?email_source=notifications&email_token=AADWVAIYP6CSYQZXV56IL73QHGOLBA5CNFSM4ILPPYYKYY3PNVWWK3TUL52HS4DFWFIHK3DMKJSXC5LFON2FEZLWNFSXPKTDN5WW2ZLOORPWSZGOCDJRJHA#discussion_r319692142>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AADWVALJN4X2J53JSWOLBULQHGOLBANCNFSM4ILPPYYA> .

angoenka · 2019-08-31T00:05:23Z

Thanks!

angoenka changed the title ~~[BEAM-7972] Always use Global window in reshuffle and then apply wind…~~ NOT FOR REVIEW [BEAM-7972] Always use Global window in reshuffle and then apply wind… Aug 14, 2019

angoenka force-pushed the fix_reshuffle branch from e003e25 to 47caba9 Compare August 14, 2019 01:51

angoenka changed the title ~~NOT FOR REVIEW [BEAM-7972] Always use Global window in reshuffle and then apply wind…~~ [BEAM-7972] Always use Global window in reshuffle and then apply wind… Aug 14, 2019

robertwb reviewed Aug 20, 2019

View reviewed changes

angoenka commented Aug 21, 2019

View reviewed changes

angoenka force-pushed the fix_reshuffle branch from 47caba9 to ec92139 Compare August 28, 2019 00:50

robertwb approved these changes Aug 30, 2019

View reviewed changes

[BEAM-7972] Always use Global window in reshuffle and then apply wind…

f22231c

…ow again.

angoenka force-pushed the fix_reshuffle branch from ec92139 to f22231c Compare August 30, 2019 22:37

angoenka merged commit d8c1146 into apache:master Aug 31, 2019

Conversation

angoenka commented Aug 13, 2019

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Uh oh!

angoenka commented Aug 13, 2019

Uh oh!

angoenka commented Aug 14, 2019

Uh oh!

y1chi commented Aug 19, 2019

Uh oh!

angoenka commented Aug 20, 2019

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

angoenka left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

angoenka commented Aug 22, 2019

Uh oh!

robertwb commented Aug 23, 2019

Uh oh!

angoenka commented Aug 24, 2019

Uh oh!

lukecwik commented Aug 24, 2019

Uh oh!

angoenka commented Aug 26, 2019

Uh oh!

angoenka commented Aug 30, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertwb commented Aug 30, 2019 via email

Uh oh!

angoenka commented Aug 31, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants