[runners-flink] #30621 use groupBy for Reshuffle in batch #30622
Conversation
|
Assigning reviewers. If you would like to opt out of this review, comment R: @shunping added as fallback since no labels match configuration Available commands:
The PR bot will only process comments in the main thread (not review comments). |
| @@ -413,12 +411,15 @@ public void translateNode( | |||
| outputType, | |||
| FlinkIdentityFunction.of(), | |||
| getCurrentTransformName(context)); | |||
| final Configuration partitionOptions = new Configuration(); | |||
| partitionOptions.setString( | |||
| Optimizer.HINT_SHIP_STRATEGY, Optimizer.HINT_SHIP_STRATEGY_REPARTITION); | |||
There was a problem hiding this comment.
Commented on the bug, but if Flink really looks at these hints, I think the implementation here is good. It is better to translate it as a hint that Flink understands than to force it.
There was a problem hiding this comment.
Ah I see from https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/dataset/transformations/#rebalance that it may just be even redistribution, which would be ignoring the hint really.
There was a problem hiding this comment.
Yes, I saw the problem appearing on a real pipeline. SDF uses Reshuffle to distribute restrictions among workers. My pipelime ended processing all restrictions on single worker, effectively detroying any parallelism.
There was a problem hiding this comment.
I just was trying to read how https://nightlies.apache.org/flink/flink-docs-release-1.13/api/java/org/apache/flink/optimizer/Optimizer.html#HINT_SHIP_STRATEGY_REPARTITION_HASH is used, it might be a good choice. But anyhow yea from what I can tell in the Flink codebase and docs, the current logic in the runner works for Redistribute.arbitrarily() but is the wrong idea for Reshuffle or Redistribute.byKey
Closes #30621
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
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, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.