[BEAM-7386] Adding EventTimeEquiJoin #15275

laraschmidt · 2021-08-03T23:15:20Z

Adding EventTimeEquiJoin which is a PTransform that joins two streams where the comparison is time-stamp bounded.

@reuvenlax @kennknowles

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.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Twister2
Go	---	---
Java
Python	---	---
XLang		---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

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

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

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

reuvenlax · 2021-08-04T22:42:21Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

+ * </pre>
+ */
+@AutoValue
+public abstract class EventTimeEquiJoin<K, V1, V2>


Would like to think on the name a bit more, as EventTimeEquiJoin seems a bit awkward to me.

I also wonder about whether this PTransform should be in terms of KVs or not. However I'm starting to think that it should be and then we have the option to wrap it in a higher-level Join transform

I looked up what Flink and Spark call them. Spark doesn't really have a name. Flink's joins seem to be different (processing on a table at a point of time instead of things being equivalent). Tyson's suggestion in the doc was EventTimeBoundedEquiJoin. Not sure I have any other suggestions. Maybe we could shorten it and move some to the function: E.g. EventTimeBoundedJoin.innerEquiOf() or something like that?

Some other options that were suggested previously:

TimestampBoundedEquijoin

EventTimeLimitedDurationEquiJoin

EventTimeScopedDurationInnerJoin

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/Pair.java

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

reuvenlax · 2021-08-04T23:21:49Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

+      Instant nextBucketStart =
+          Instant.ofEpochMilli(
+              cleanupTime.getMillis() / TIMER_BUCKET * TIMER_BUCKET + TIMER_BUCKET);
+      cleanupTimers.get(Long.toString(nextBucketStart.getMillis())).set(nextBucketStart);


ideally we should call withOutputTimestamp to hold the watermark back to the earliest buffered element in that timer's range. Doing this probably requires keeping a histogram in a ValueState tracking the min timestamp per minute bucket.

reuvenlax · 2021-08-12T17:11:51Z

Let me know when you're ready for another review round!

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

laraschmidt · 2021-08-20T23:25:46Z

@reuvenlax @kennknowles I implemented changes as if we were to remove the checks we discussed offline (one I avoided with the skew and the other by removing it). Please take a look, thanks!

laraschmidt · 2021-09-07T20:38:20Z

Ping. :) @reuvenlax @kennknowles

aaltay · 2021-09-16T17:21:30Z

@kennknowles @reuvenlax - Could you please take a look at this?

aaltay · 2021-09-30T20:24:01Z

@kennknowles @reuvenlax - Could you please take a look at this?

@laraschmidt - Could you look at the failing precommit test or re-run it?

laraschmidt · 2021-10-04T17:57:25Z

Run Java PreCommit

laraschmidt · 2021-10-04T18:05:39Z

Run Java_Examples_Dataflow PreCommit

laraschmidt · 2021-10-04T21:04:22Z

Run Java_Examples_Dataflow PreCommit

laraschmidt · 2021-10-04T21:04:34Z

Run Java PreCommit

kennknowles

Nice

kennknowles · 2021-10-11T17:57:55Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/PairCoder.java

+  @Override
+  public void encode(Pair<V1, V2> value, OutputStream outStream)
+      throws CoderException, IOException {
+    firstCoder.encode(value.getFirst(), outStream);


Have we deprecated the coder "context" idea? Or do you just not want to apply it here? I would expect PairCoder to be identical to KvCoder, anyhow.

Not familiar with the deprecated context thing but I copied it over from KVCoder. What exactly is it used for? I assume that in my case I want context on both (whereas the KV coder only put it on the value).

Hrm, having both failed so I swapped it to just be on the second element.

kennknowles · 2021-10-11T20:38:58Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

+        Instant ts,
+        OrderedListState<ThisT> thisCollection,
+        OrderedListState<OtherT> otherCollection,
+        ValueState<Instant> oldestTimestampState,


I can see how having a specific timestamp queue state would make it easier to have all the methods we need in a single "state type". Interestingly enough, the original design for state (before all the annotations) made it pretty easy to define composite types of state, whereas now we really cannot.

kennknowles · 2021-10-11T20:41:30Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java

+ * @param <V2> the type of the value in the second {@code PCollection}
+ * </pre>
+ */
+@AutoValue


@Experimental

laraschmidt · 2021-10-12T19:03:19Z

Addressed comments, PTAL. :) Thanks! @kennknowles

aaltay · 2021-10-22T01:22:23Z

@kennknowles - Could you please take a look?

aaltay · 2021-12-29T18:23:02Z

~~Is this still relevant? Should we close it?~~

Ignore that. I saw the recent changes.

reuvenlax · 2021-12-29T18:33:03Z

I think it's still relevant. It was blocked on Lara's other PR.

…

On Wed, Dec 29, 2021 at 10:23 AM Ahmet Altay ***@***.***> wrote: Is this still relevant? Should we close it? — Reply to this email directly, view it on GitHub <#15275 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVLMOFDSNGLDW7JIHBTUTNGZDANCNFSM5BPZWL6Q> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>

aaltay · 2022-01-07T04:44:05Z

@laraschmidt - how is your other PR doing? :)

laraschmidt · 2022-01-07T17:10:43Z

The other one is in but it does not work for portable runner because making it work for protable runner breaks dataflow on portable runner. :)

But the changes can actually be treated as pretty independent. That one was making the function we use here not as deprecated but we can still use it in this PR even if deprecated. So this is still ready for review.

reuvenlax · 2022-01-07T17:54:54Z

But you needed to make that function apply to timers as well, right?

…

On Fri, Jan 7, 2022 at 9:10 AM laraschmidt ***@***.***> wrote: The other one is in but it does not work for portable runner because making it work for protable runner breaks dataflow on portable runner. :) But the changes can actually be treated as pretty independent. That one was making the function we use here not as deprecated but we can still use it in this PR even if deprecated. So this is still ready for review. — Reply to this email directly, view it on GitHub <#15275 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVPVQDONYV2V4STERPTUU4NB5ANCNFSM5BPZWL6Q> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>

laraschmidt · 2022-01-07T18:23:28Z

That was for an earlier iteration. The solution we ended up going with only outputs from processElement and only uses timers to hold the watermark. So we didn't end up actually needing to allow older elements from timers because we accepted that we will allow older elements from processElement function.

reuvenlax · 2022-01-07T18:24:44Z

Right, but you need to be able to set a timer with a hold older than the current input element, right?

…

On Fri, Jan 7, 2022 at 10:23 AM laraschmidt ***@***.***> wrote: That was for an earlier iteration. The solution we ended up going with only outputs from the actual doFn and only uses timers to hold the watermark. So we didn't end up actually needing to allow older elements from timers because we accepted that we will allow older elements from the actual DoFn. — Reply to this email directly, view it on GitHub <#15275 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVIA3SCSRCB5NQQW47TUU4VSVANCNFSM5BPZWL6Q> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>

laraschmidt · 2022-01-07T18:35:28Z

I'm trying to page this back in. But if I remember correctly, the first element holds the watermark back until all not-late second elements would have appeared. So we don't actually need to hold it backwards. It's just the processElement that puts out older elements.

Here's the relevant code for the timer which are not negative:
if (elementAffectsWatermarkHolds) {
addTimer(watermarkHolds, ts.plus(thisCollectionValidFor)).withOutputTimestamp(ts);

reuvenlax · 2022-01-07T19:22:36Z

I need to take a look at the code, since it's changed a lot. However elements don't arrive in timestamp order, so I'm not sure how you would avoid this.

…

On Fri, Jan 7, 2022 at 10:35 AM laraschmidt ***@***.***> wrote: I'm trying to page this back in. But if I remember correctly, the first element holds the watermark back until all not-late second elements would have appeared. So we don't actually need to hold it backwards. It's just the processElement that puts out older elements. Here's the relevant code for the timer which are not negative: if (elementAffectsWatermarkHolds) { addTimer(watermarkHolds, ts.plus(thisCollectionValidFor)).withOutputTimestamp(ts); — Reply to this email directly, view it on GitHub <#15275 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVOZTULKXCPAB2WT3VTUU4W7ZANCNFSM5BPZWL6Q> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>

laraschmidt · 2022-01-07T23:41:55Z

We use the same watermark hold for a bucket of elements (E.g. arriving between t and t+e). And we just hold it back til t+e. So it shouldn't matter that they come out of order from what I can tell.

aaltay · 2022-01-21T00:46:11Z

What is the next step on this PR? - (I will stop pinging. Let me know if I can do anything to help.)

github-actions · 2022-03-22T12:52:31Z

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

github-actions · 2022-03-30T12:52:36Z

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

reuvenlax self-requested a review August 4, 2021 21:44

reuvenlax reviewed Aug 4, 2021

View reviewed changes

laraschmidt force-pushed the equijoin branch from 5d88f17 to 3569a2d Compare August 5, 2021 21:30

reuvenlax reviewed Aug 12, 2021

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/join/EventTimeEquiJoin.java Outdated Show resolved Hide resolved

laraschmidt force-pushed the equijoin branch from 3569a2d to d1d605f Compare August 20, 2021 23:22

laraschmidt force-pushed the equijoin branch 6 times, most recently from ec252da to 7a8ba6d Compare August 24, 2021 23:35

kennknowles self-requested a review September 2, 2021 20:32

laraschmidt force-pushed the equijoin branch 3 times, most recently from a17efd4 to 9c8eca7 Compare September 20, 2021 21:46

laraschmidt force-pushed the equijoin branch from 9c8eca7 to 4102227 Compare October 1, 2021 18:49

kennknowles reviewed Oct 11, 2021

View reviewed changes

laraschmidt force-pushed the equijoin branch 2 times, most recently from be5bf5c to 08fc756 Compare October 12, 2021 19:02

laraschmidt added 2 commits December 16, 2021 15:31

Adding EventTimeEquiJoin and test.

95cee94

Addressing review feedback and fixing watermark handling.

ddaf28b

laraschmidt force-pushed the equijoin branch from 08fc756 to ddaf28b Compare December 16, 2021 23:33

github-actions bot added the stale label Mar 22, 2022

github-actions bot closed this Mar 30, 2022

[BEAM-7386] Adding EventTimeEquiJoin #15275

[BEAM-7386] Adding EventTimeEquiJoin #15275

Conversation

laraschmidt commented Aug 3, 2021 • edited

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reuvenlax commented Aug 12, 2021

laraschmidt commented Aug 20, 2021

laraschmidt commented Sep 7, 2021

aaltay commented Sep 16, 2021

aaltay commented Sep 30, 2021

laraschmidt commented Oct 4, 2021

laraschmidt commented Oct 4, 2021

laraschmidt commented Oct 4, 2021

laraschmidt commented Oct 4, 2021

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

laraschmidt commented Oct 12, 2021

aaltay commented Oct 22, 2021

aaltay commented Dec 29, 2021 • edited

reuvenlax commented Dec 29, 2021 via email

aaltay commented Jan 7, 2022

laraschmidt commented Jan 7, 2022

reuvenlax commented Jan 7, 2022 via email

laraschmidt commented Jan 7, 2022 • edited

reuvenlax commented Jan 7, 2022 via email

laraschmidt commented Jan 7, 2022

reuvenlax commented Jan 7, 2022 via email

laraschmidt commented Jan 7, 2022

aaltay commented Jan 21, 2022

github-actions bot commented Mar 22, 2022

github-actions bot commented Mar 30, 2022

laraschmidt commented Aug 3, 2021 •

edited

`ValidatesRunner` compliance status (on master branch)

aaltay commented Dec 29, 2021 •

edited

laraschmidt commented Jan 7, 2022 •

edited