You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "alxmrs (via GitHub)" <gi...@apache.org> on 2024/10/24 22:15:21 UTC
[PR] Windowing Support for the Dask Runner [beam]
alxmrs opened a new pull request, #32941:
URL: https://github.com/apache/beam/pull/32941
This CL adds basic Windowing support to this runner, including a few tests for side inputs – take three (#27618, #23913).
CC: @cisaacstern
Reviewers: @jrmccluskey
------------------------
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2477789769
That sounds good. Hey, @jacobtomlinson: by any chance, do you have cycles to review this PR, focusing on the Dask aspect of the code? This one has been a long time coming! I would appreciate any and all help you can offer (and no pressure if time is sparse).
--
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] Windowing Support for the Dask Runner [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2457014616
Reminder, please take a look at this pr: @jrmccluskey @Abacn
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1819326027
##########
.github/workflows/dask_runner_tests.yml:
##########
@@ -22,7 +22,7 @@ name: Dask Runner Tests
on:
schedule:
- cron: '3 7 * * *'
- pull_request:
+ pull_request_target:
Review Comment:
change to "pull_request_target" will make the workflow runs on HEAD instead of the PR branch. Is there any issue using "pull_request" trigger?
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2461675716
@Abacn May I have some help getting the windows build to work? It looks tractable:
```
C:\hostedtoolcache\windows\Python\3.10.11\x64\python.exe: can't open file
'D:\\a\\beam\\beam\\sdks\\python\\target\\.tox\\py310-win-dask\\Scripts\\pip': [Errno 2] No such file or directory
```
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2461766171
Ok, I think I got it working on windows, too.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1829564660
##########
sdks/python/tox.ini:
##########
@@ -109,9 +109,15 @@ commands =
bash {toxinidir}/scripts/run_pytest.sh {envname} "{posargs}"
[testenv:py{39,310,311,312}-dask]
-extras = test,dask
+extras = test,dask,dataframes
+commands_pre =
+ # FIXME(cisaacstern): remove this block before merge. Adding for now to get tests to pass in
+ # CI during development. See related note in setup.py; `distributed` needs to be installed
+ # before `dask` for development env to work correctly.
+ pip install 'distributed==2023.12.1'
+ pip install -U 'dask @ git+https://github.com/cisaacstern/dask.git@tokenize-bag-groupby-key'
Review Comment:
"remove this block before merge"
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1829567747
##########
sdks/python/apache_beam/runners/dask/dask_runner.py:
##########
@@ -31,12 +31,22 @@
from apache_beam.pipeline import PipelineVisitor
from apache_beam.runners.dask.overrides import dask_overrides
from apache_beam.runners.dask.transform_evaluator import TRANSLATIONS
+from apache_beam.runners.dask.transform_evaluator import DaskBagWindowedIterator
+from apache_beam.runners.dask.transform_evaluator import Flatten
from apache_beam.runners.dask.transform_evaluator import NoOp
from apache_beam.runners.direct.direct_runner import BundleBasedDirectRunner
from apache_beam.runners.runner import PipelineResult
from apache_beam.runners.runner import PipelineState
+from apache_beam.transforms.sideinputs import SideInputMap
from apache_beam.utils.interactive_utils import is_in_notebook
+try:
+ # Added to try to prevent threading related issues, see
+ # https://github.com/pytest-dev/pytest/issues/3216#issuecomment-1502451456
+ import dask.distributed as ddist
+except ImportError:
+ distributed = {}
Review Comment:
Please check is this intended? `distributed = {}` this just creates a local variable of empty dict.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "jacobtomlinson (via GitHub)" <gi...@apache.org>.
jacobtomlinson commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1846191039
##########
sdks/python/apache_beam/runners/dask/dask_runner.py:
##########
@@ -99,8 +108,16 @@ def wait_until_finish(self, duration=None) -> str:
if duration is not None:
# Convert milliseconds to seconds
duration /= 1000
- self.client.wait_for_workers(timeout=duration)
- self.client.gather(self.futures, errors='raise')
+ for _ in ddist.as_completed(self.futures,
+ timeout=duration,
+ with_results=True):
+ # without gathering results, worker errors are not raised on the client:
+ # https://distributed.dask.org/en/stable/resilience.html#user-code-failures
+ # so we want to gather results to raise errors client-side, but we do
+ # not actually need to use the results here, so we just pass. to gather,
+ # we use the iterative `as_completed(..., with_results=True)`, instead
+ # of aggregate `client.gather`, to minimize memory footprint of results.
+ pass
Review Comment:
Good to know. Sounds like `as_completed` is the way then!
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2483942711
Thank you, merging for 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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #32941:
URL: https://github.com/apache/beam/pull/32941
--
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] Windowing Support for the Dask Runner [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2436517445
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @jrmccluskey for label python.
R: @Abacn for label build.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1820038870
##########
.github/workflows/dask_runner_tests.yml:
##########
@@ -62,7 +62,7 @@ jobs:
strategy:
fail-fast: false
matrix:
- os: [ubuntu-latest, macos-latest, windows-latest]
Review Comment:
Ah, something was lost in the shuffle: https://github.com/apache/beam/pull/27618/files#r1271370322
IIRC writing this so long ago, a dependency (with dask?) caused windows to fail. However, this may have “resolved itself” given how much the code has changed.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1844297026
##########
sdks/python/apache_beam/runners/dask/dask_runner.py:
##########
@@ -99,8 +108,16 @@ def wait_until_finish(self, duration=None) -> str:
if duration is not None:
# Convert milliseconds to seconds
duration /= 1000
- self.client.wait_for_workers(timeout=duration)
- self.client.gather(self.futures, errors='raise')
+ for _ in ddist.as_completed(self.futures,
+ timeout=duration,
+ with_results=True):
+ # without gathering results, worker errors are not raised on the client:
+ # https://distributed.dask.org/en/stable/resilience.html#user-code-failures
+ # so we want to gather results to raise errors client-side, but we do
+ # not actually need to use the results here, so we just pass. to gather,
+ # we use the iterative `as_completed(..., with_results=True)`, instead
+ # of aggregate `client.gather`, to minimize memory footprint of results.
+ pass
Review Comment:
I just tested this. Replacing the loop with:
```
ddist.wait(self.futures, timeout=duration)
```
fails a test as it seems the runner swallows errors. With `as_completed()`, it looks like the errors propagate to the right place.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "jacobtomlinson (via GitHub)" <gi...@apache.org>.
jacobtomlinson commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1843462861
##########
sdks/python/apache_beam/runners/dask/dask_runner.py:
##########
@@ -99,8 +108,16 @@ def wait_until_finish(self, duration=None) -> str:
if duration is not None:
# Convert milliseconds to seconds
duration /= 1000
- self.client.wait_for_workers(timeout=duration)
- self.client.gather(self.futures, errors='raise')
+ for _ in ddist.as_completed(self.futures,
+ timeout=duration,
+ with_results=True):
+ # without gathering results, worker errors are not raised on the client:
+ # https://distributed.dask.org/en/stable/resilience.html#user-code-failures
+ # so we want to gather results to raise errors client-side, but we do
+ # not actually need to use the results here, so we just pass. to gather,
+ # we use the iterative `as_completed(..., with_results=True)`, instead
+ # of aggregate `client.gather`, to minimize memory footprint of results.
+ pass
Review Comment:
Would `ddist.wait(self.futures, ...` achieve the same result without having to iterate and `pass`?
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2463667219
I think the failing tests are just flaky.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1831976069
##########
sdks/python/apache_beam/runners/dask/dask_runner.py:
##########
@@ -31,12 +31,22 @@
from apache_beam.pipeline import PipelineVisitor
from apache_beam.runners.dask.overrides import dask_overrides
from apache_beam.runners.dask.transform_evaluator import TRANSLATIONS
+from apache_beam.runners.dask.transform_evaluator import DaskBagWindowedIterator
+from apache_beam.runners.dask.transform_evaluator import Flatten
from apache_beam.runners.dask.transform_evaluator import NoOp
from apache_beam.runners.direct.direct_runner import BundleBasedDirectRunner
from apache_beam.runners.runner import PipelineResult
from apache_beam.runners.runner import PipelineState
+from apache_beam.transforms.sideinputs import SideInputMap
from apache_beam.utils.interactive_utils import is_in_notebook
+try:
+ # Added to try to prevent threading related issues, see
+ # https://github.com/pytest-dev/pytest/issues/3216#issuecomment-1502451456
+ import dask.distributed as ddist
+except ImportError:
+ distributed = {}
Review Comment:
Ah, on closer look -- you're right. It looks like it should be `ddist = {}`.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1831976650
##########
.github/workflows/dask_runner_tests.yml:
##########
@@ -62,7 +62,7 @@ jobs:
strategy:
fail-fast: false
matrix:
- os: [ubuntu-latest, macos-latest, windows-latest]
Review Comment:
Adding back windows for now; let's see if it passes tests.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2468414388
@Abacn, could you take another look on @alxmrs 's changes?
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2477316950
Sorry for late response. My comments were most on the infrastructure / testing related components. For that part LGTM. For dask runner implementation would it be possible to find a reviewer who knows Dask also?
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1820037932
##########
.github/workflows/dask_runner_tests.yml:
##########
@@ -22,7 +22,7 @@ name: Dask Runner Tests
on:
schedule:
- cron: '3 7 * * *'
- pull_request:
+ pull_request_target:
Review Comment:
No issue at all, happy to change it.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1831973785
##########
sdks/python/tox.ini:
##########
@@ -109,9 +109,15 @@ commands =
bash {toxinidir}/scripts/run_pytest.sh {envname} "{posargs}"
[testenv:py{39,310,311,312}-dask]
-extras = test,dask
+extras = test,dask,dataframes
+commands_pre =
+ # FIXME(cisaacstern): remove this block before merge. Adding for now to get tests to pass in
+ # CI during development. See related note in setup.py; `distributed` needs to be installed
+ # before `dask` for development env to work correctly.
+ pip install 'distributed==2023.12.1'
+ pip install -U 'dask @ git+https://github.com/cisaacstern/dask.git@tokenize-bag-groupby-key'
Review Comment:
Thanks for catching 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
Re: [PR] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2479677786
Hey @Abacn! In light of Jacob's review, do you think we're good-to-go? :)
--
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] Windowing Support for the Dask Runner [beam]
Posted by "alxmrs (via GitHub)" <gi...@apache.org>.
alxmrs commented on PR #32941:
URL: https://github.com/apache/beam/pull/32941#issuecomment-2475843063
Hey there! I'd love to get this merged. Please let me know if there is any more I can do.
--
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] Windowing Support for the Dask Runner [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #32941:
URL: https://github.com/apache/beam/pull/32941#discussion_r1819326600
##########
.github/workflows/dask_runner_tests.yml:
##########
@@ -62,7 +62,7 @@ jobs:
strategy:
fail-fast: false
matrix:
- os: [ubuntu-latest, macos-latest, windows-latest]
Review Comment:
why windows is removed?
--
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