You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Abacn (via GitHub)" <gi...@apache.org> on 2023/01/25 21:22:29 UTC
[GitHub] [beam] Abacn opened a new pull request, #25173: Fix Jdbc Write after window assigned
Abacn opened a new pull request, #25173:
URL: https://github.com/apache/beam/pull/25173
Fixes #21882
Tested on flink runner (v1.14)
- v2.44.0 failed with mentioned exception
- with the change, suceeded and records written
unfortunately it is not possible to unit test the change because the direct runner apparently does not do the window-coder things.
* Assign last seen window to the remaining batch in finishBundle
**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:
- [ ] 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://beam.apache.org/contribute/get-started-contributing/#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.
--
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
[GitHub] [beam] pabloem commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1409017185
lgtm
--
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
[GitHub] [beam] Abacn commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1404240912
R: @pabloem
--
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
[GitHub] [beam] pabloem commented on a diff in pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on code in PR #25173:
URL: https://github.com/apache/beam/pull/25173#discussion_r1087318254
##########
sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java:
##########
@@ -1716,31 +1714,10 @@ static <T> PCollection<Iterable<T>> batchElements(
.apply(Values.create());
} else {
iterables =
- input.apply(
- ParDo.of(
- new DoFn<T, Iterable<T>>() {
- @Nullable List<T> outputList;
-
- @ProcessElement
- public void process(ProcessContext c) {
- if (outputList == null) {
- outputList = new ArrayList<>();
- }
- outputList.add(c.element());
- if (outputList.size() > batchSize) {
- c.output(outputList);
- outputList = null;
- }
- }
-
- @FinishBundle
- public void finish(FinishBundleContext c) {
- if (outputList != null && outputList.size() > 0) {
- c.output(outputList, Instant.now(), GlobalWindow.INSTANCE);
- }
- outputList = null;
- }
- }));
+ input
+ .apply(WithKeys.<String, T>of(""))
+ .apply(GroupIntoBatches.<String, T>ofSize(batchSize).withShardedKey())
Review Comment:
```suggestion
.apply(GroupIntoBatches.<String, T>ofSize(batchSize)..withMaxBufferingDuration(Duration.millis(200)))
```
--
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
[GitHub] [beam] Abacn commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1406968467
PTAL @pabloem 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
[GitHub] [beam] pabloem commented on a diff in pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on code in PR #25173:
URL: https://github.com/apache/beam/pull/25173#discussion_r1087318363
##########
sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java:
##########
@@ -1716,31 +1714,10 @@ static <T> PCollection<Iterable<T>> batchElements(
.apply(Values.create());
} else {
iterables =
- input.apply(
- ParDo.of(
- new DoFn<T, Iterable<T>>() {
- @Nullable List<T> outputList;
-
- @ProcessElement
- public void process(ProcessContext c) {
- if (outputList == null) {
- outputList = new ArrayList<>();
- }
- outputList.add(c.element());
- if (outputList.size() > batchSize) {
- c.output(outputList);
- outputList = null;
- }
- }
-
- @FinishBundle
- public void finish(FinishBundleContext c) {
- if (outputList != null && outputList.size() > 0) {
- c.output(outputList, Instant.now(), GlobalWindow.INSTANCE);
- }
- outputList = null;
- }
- }));
+ input
+ .apply(WithKeys.<String, T>of(""))
+ .apply(GroupIntoBatches.<String, T>ofSize(batchSize).withShardedKey())
Review Comment:
this creates the same kind of per-batch grouping but no autosharding
--
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
[GitHub] [beam] pabloem commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1404398659
Autosharding is a particular capability of the Dataflow runner. See: https://cloud.google.com/blog/products/data-analytics/3x-dataflow-throughput-auto-sharding-bigquery
It alows us to dynamically grow the number of individual shards by the number of streaming workers, instead of having a fixed number of shards, which is the case for other runners. The outcome is that
- Without autosharding: for a given key X, all of the elements for that key will end up processed serially (usually by the same worker)
- With autosharding, GroupIntoBatches uses a ShardedKey where, for a key X, elements for that key will be batched into parallel shards that grow proportionally to the number of workers
--
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
[GitHub] [beam] Abacn commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1409037650
@pabloem yes, tested locally on flink runner (direct runner did not reproduce error at all; flink can reproduce before the change and no error after the change)
--
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
[GitHub] [beam] Abacn commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1404277885
@pabloem thanks for the input. Would you mind reminding me where does "auto"sharding taking effect, to me it looks like both branch are sharding incoming elements into batch of size batchSize. i.e. what is the difference with/without withAutoSharding?
--
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
[GitHub] [beam] Abacn commented on a diff in pull request #25173: Fix Jdbc Write after window assigned
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #25173:
URL: https://github.com/apache/beam/pull/25173#discussion_r1087325840
##########
sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java:
##########
@@ -1716,31 +1714,10 @@ static <T> PCollection<Iterable<T>> batchElements(
.apply(Values.create());
} else {
iterables =
- input.apply(
- ParDo.of(
- new DoFn<T, Iterable<T>>() {
- @Nullable List<T> outputList;
-
- @ProcessElement
- public void process(ProcessContext c) {
- if (outputList == null) {
- outputList = new ArrayList<>();
- }
- outputList.add(c.element());
- if (outputList.size() > batchSize) {
- c.output(outputList);
- outputList = null;
- }
- }
-
- @FinishBundle
- public void finish(FinishBundleContext c) {
- if (outputList != null && outputList.size() > 0) {
- c.output(outputList, Instant.now(), GlobalWindow.INSTANCE);
- }
- outputList = null;
- }
- }));
+ input
+ .apply(WithKeys.<String, T>of(""))
+ .apply(GroupIntoBatches.<String, T>ofSize(batchSize).withShardedKey())
Review Comment:
thanks for the detailed explanation. I am wondering what is the performance impact of changing it to GroupIntoBatches. Should we preserve the current implementation (a DoFn) used for batch, and use GroupIntoBatches in streaming with/without autosharding?
--
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
[GitHub] [beam] pabloem commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1409017702
have you tested this?
--
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
[GitHub] [beam] pabloem merged pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem merged PR #25173:
URL: https://github.com/apache/beam/pull/25173
--
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
[GitHub] [beam] pabloem commented on pull request #25173: Fix Jdbc Write after window assigned
Posted by "pabloem (via GitHub)" <gi...@apache.org>.
pabloem commented on PR #25173:
URL: https://github.com/apache/beam/pull/25173#issuecomment-1404256858
ah I understand the issue now. Elements in a bundle can come from multiple windows. I would say that ... a fix could instead rely on `GroupIntoBatches` for both paths, and only add autosharding on the first one?
thoughts?
fyi @reuvenlax
--
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