You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "twosom (via GitHub)" <gi...@apache.org> on 2024/11/25 11:19:52 UTC
[PR] Add OrderedListState support for SparkRunner [beam]
twosom opened a new pull request, #33212:
URL: https://github.com/apache/beam/pull/33212
**Please** add a meaningful description for your change here
fixes #33211
fixes #31724
fixes #31723
This PR contains these changes
- added `OrderedList` in `SparkStateInternals`
- implementation `equals` and `hashcode` in `FlinkOrderedListState`
- added `OrderedListState` test
- fixed `FlinkRunner` state test
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2508008261
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2520463508
Please check this:
:runners:spark:3:analyzeClassesDependencies FAILED
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2518379364
R: @shunping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2514827626
Run Flink Container PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2552502032
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892180756
##########
CHANGES.md:
##########
@@ -71,11 +71,15 @@
## New Features / Improvements
+* Improved batch performance of SparkRunner's GroupByKey ([#20943](https://github.com/apache/beam/pull/20943)).
+* Support OnWindowExpiration in Prism ([#32211](https://github.com/apache/beam/issues/32211)).
Review Comment:
Ah! My mistake. This happened while resolving git conflicts through the web UI.😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892198117
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
Review Comment:
Got it!
I'll clean up the commit history by merging formatting changes into meaningful commits and keeping independent changes separate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2497846477
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2508004138
Run PreCommit Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2552501603
Run Java_GCP_IO_Direct PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2508009230
Run Flink Container PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2513950914
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2518511197
Did you rebase your branch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2520546051
@liferoad
Sorry, my bad. Fixed the import statement to use Beam's vendored Guava Lists instead of clearspring's.
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892259483
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final FlinkOrderedListState<?> that = (FlinkOrderedListState<?>) o;
+
+ return namespace.equals(that.namespace) && stateId.equals(that.stateId);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = namespace.hashCode();
Review Comment:
```java
@Override
public int hashCode() {
int result = Objects.hashCode(namespace);
result = 31 * result + Objects.hashCode(stateId);
return result;
}
```
Thank you. Would this change be appropriate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1894007388
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final FlinkOrderedListState<?> that = (FlinkOrderedListState<?>) o;
+
+ return namespace.equals(that.namespace) && stateId.equals(that.stateId);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = namespace.hashCode();
Review Comment:
@kennknowles
The current implementation follows the same hashCode semantics used in FlinkStateInternals. I'm a bit unclear whether suggesting `Objects.hashCode` means we should replace the current implementation, or if you're suggesting that overriding `hashCode` is unnecessary altogether.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2557211262
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2519427065
Run Java_Spark3_Versions PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2513757747
retest this please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892259483
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final FlinkOrderedListState<?> that = (FlinkOrderedListState<?>) o;
+
+ return namespace.equals(that.namespace) && stateId.equals(that.stateId);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = namespace.hashCode();
Review Comment:
```java
@Override
public int hashCode() {
int result = Objects.hashCode(namespace);
result = 31 * result + Objects.hashCode(stateId);
return result;
}
```
Thank you. Would this change be appropriate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1894416064
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java:
##########
@@ -622,4 +627,82 @@ public Boolean read() {
};
}
}
+
+ private final class SparkOrderedListState<T> extends AbstractState<List<TimestampedValue<T>>>
+ implements OrderedListState<T> {
+
+ private SparkOrderedListState(StateNamespace namespace, String id, Coder<T> coder) {
+ super(namespace, id, ListCoder.of(TimestampedValue.TimestampedValueCoder.of(coder)));
+ }
+
+ private SortedMap<Instant, TimestampedValue<T>> readAsMap() {
+ final List<TimestampedValue<T>> listValues =
Review Comment:
I see. If Flink is implemented then it is OK with me to follow that precedent. My point was that this does not actually add capability that is more than `ValueState` provides. It is just a minor API wrapper adjustment - still useful but not the main purpose.
So we can merge with this design. But if you think about following up, here is how we would really like this to behave:
- `add` should call some native Spark API that writes the element without reading the list
- `readRange` should only read the requested range, ideally seeking in near-constant time (aka without a scan or sort)
- `clearRange` should also seek in near-constant time
- `isEmpty` should not read the list
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final FlinkOrderedListState<?> that = (FlinkOrderedListState<?>) o;
+
+ return namespace.equals(that.namespace) && stateId.equals(that.stateId);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = namespace.hashCode();
Review Comment:
Overriding equality and hashcode should always occur together, so it is necessary to override `hashCode`.
I was just suggesting this pattern instead of doing your own math: https://github.com/apache/beam/blob/7c86bf3103fec1d3a6a3a3cbe6ace8a14a85b723/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java#L88
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892115490
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final FlinkOrderedListState<?> that = (FlinkOrderedListState<?>) o;
+
+ return namespace.equals(that.namespace) && stateId.equals(that.stateId);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = namespace.hashCode();
Review Comment:
Can use `Objects.hashCode`
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java:
##########
@@ -571,6 +573,27 @@ public void clear() {
throw new RuntimeException("Error clearing state.", e);
}
}
+
+ @Override
Review Comment:
This change is nice to have. It should be in a separate commit. Can you please adjust the commit history so that each independent change is in a clearly labeled commit, and I will not squash them. In other words, merge the spotless chore and other non-meaningful commits into the other commits so that the commits in the PR are each meaningful on their own.
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java:
##########
@@ -622,4 +627,82 @@ public Boolean read() {
};
}
}
+
+ private final class SparkOrderedListState<T> extends AbstractState<List<TimestampedValue<T>>>
+ implements OrderedListState<T> {
+
+ private SparkOrderedListState(StateNamespace namespace, String id, Coder<T> coder) {
+ super(namespace, id, ListCoder.of(TimestampedValue.TimestampedValueCoder.of(coder)));
+ }
+
+ private SortedMap<Instant, TimestampedValue<T>> readAsMap() {
+ final List<TimestampedValue<T>> listValues =
Review Comment:
The reason for each additional kind of state is to efficiently offer a novel form of a state access. The state access here as the same performance characteristics as `ValueState`. It is actually better for the runner to reject a pipeline than to run it with performance characteristics that don't match the expected performance contract.
Is there some underlying mechanism in Spark that could implement OrderedListState efficiently and scalably?
##########
CHANGES.md:
##########
@@ -71,11 +71,15 @@
## New Features / Improvements
+* Improved batch performance of SparkRunner's GroupByKey ([#20943](https://github.com/apache/beam/pull/20943)).
+* Support OnWindowExpiration in Prism ([#32211](https://github.com/apache/beam/issues/32211)).
Review Comment:
Some of the changes listed here don't seem to be part of this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2543465770
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2678316861
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2513984709
Run Flink Container PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2498032085
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/33212?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 59.01%. Comparing base [(`e5bd69d`)](https://app.codecov.io/gh/apache/beam/commit/e5bd69db5bf9258021bb6c7a80e84441f0159c2c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`74033fa`)](https://app.codecov.io/gh/apache/beam/commit/74033fa7510ff415b3b49fa3e3f05185b4743974?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #33212 +/- ##
============================================
+ Coverage 57.42% 59.01% +1.58%
- Complexity 1475 3162 +1687
============================================
Files 970 1136 +166
Lines 154525 175078 +20553
Branches 1076 3354 +2278
============================================
+ Hits 88743 103324 +14581
- Misses 63578 68406 +4828
- Partials 2204 3348 +1144
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/33212/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/33212/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.27% <ø> (+1.69%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/33212?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
----
🚨 Try these New Features:
- [Flaky Tests Detection](https://docs.codecov.com/docs/test-result-ingestion-beta) - Detect and resolve failed and flaky tests
- [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis) - Avoid shipping oversized bundles
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2518381628
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2519032129
> Did you rebase your branch?
Thanks!
i rebased now!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on PR #33212:
URL: https://github.com/apache/beam/pull/33212#issuecomment-2520653072
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1892239162
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java:
##########
@@ -622,4 +627,82 @@ public Boolean read() {
};
}
}
+
+ private final class SparkOrderedListState<T> extends AbstractState<List<TimestampedValue<T>>>
+ implements OrderedListState<T> {
+
+ private SparkOrderedListState(StateNamespace namespace, String id, Coder<T> coder) {
+ super(namespace, id, ListCoder.of(TimestampedValue.TimestampedValueCoder.of(coder)));
+ }
+
+ private SortedMap<Instant, TimestampedValue<T>> readAsMap() {
+ final List<TimestampedValue<T>> listValues =
Review Comment:
I agree with your point. Let me share my thoughts on why I chose this implementation.
I've noticed that ListState/OrderedListState is mostly used in situations where writes happen much more frequently than reads. That's why I went with ArrayList instead of SortedMap - it's simply better at handling these frequent writes.
When it comes to reading data, it usually happens in just a couple of scenarios - either during OnTimer execution or when the list hits a certain size. So even if the read performance takes a small hit, it's not really going to affect the overall performance much.
It's also worth mentioning that FlinkOrderedListState uses the same approach, which gives me confidence in this design choice.
That's why I think the current implementation makes more sense for real-world usage patterns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom commented on code in PR #33212:
URL: https://github.com/apache/beam/pull/33212#discussion_r1897383414
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkStateInternals.java:
##########
@@ -622,4 +627,82 @@ public Boolean read() {
};
}
}
+
+ private final class SparkOrderedListState<T> extends AbstractState<List<TimestampedValue<T>>>
+ implements OrderedListState<T> {
+
+ private SparkOrderedListState(StateNamespace namespace, String id, Coder<T> coder) {
+ super(namespace, id, ListCoder.of(TimestampedValue.TimestampedValueCoder.of(coder)));
+ }
+
+ private SortedMap<Instant, TimestampedValue<T>> readAsMap() {
+ final List<TimestampedValue<T>> listValues =
Review Comment:
@kennknowles
Hello. I've been reviewing the Spark documentation over the weekend.
Unfortunately, as far as I can tell, Spark doesn't have the native state API that we're looking for. Moreover, since the current SparkStateInternals is a class implemented independently of the Spark API, it's also difficult to apply RDD-based APIs.
The performance requirements you mentioned seem to be well satisfied by `WindmillOrderedList`. Given that the current implementation cannot guarantee the same performance, I completely understand if the PR doesn't get merged.
However, regardless of whether this PR gets merged or not, I will continue researching to meet the requirements you've outlined. I plan to explore better implementation approaches, taking inspiration from concepts like pendingAdds and pendingDeletes as used in `WindmillOrderedList`.
Belated Merry Christmas.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add OrderedListState support for SparkRunner [beam]
Posted by "twosom (via GitHub)" <gi...@apache.org>.
twosom closed pull request #33212: Add OrderedListState support for SparkRunner
URL: https://github.com/apache/beam/pull/33212
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org