Eliaaazzz opened a new pull request, #37965:
URL: https://github.com/apache/beam/pull/37965
Fixes #24422.
Python `iobase.Read` uses a Splittable DoFn internally. PortableRunner's
default batch pre-optimization path did not include `translations.expand_sdf`,
so portable runners that do not support SDFs natively can receive a single SDF
`ParDo` for reads. In Spark, that means Python reads such as `ReadFromParquet`
may execute on a single partition without parallelization.
This change:
- adds `translations.expand_sdf` to PortableRunner's default optimization
phases
- allows `--experiments=pre_optimize=expand_sdf` as an explicit custom phase
- adds optimizer coverage for:
- default SDF expansion
- explicit `pre_optimize=expand_sdf`
- `beam.io.Read(BoundedSource)` expansion, including the expected
`RESHUFFLE`
The bounded read test covers the issue scenario behind `ReadFromParquet` and
similar Python `Read` transforms by verifying that the optimized pipeline
contains the SDF component stages:
- `PAIR_WITH_RESTRICTION`
- `SPLIT_AND_SIZE_RESTRICTIONS`
- `PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS`
- `RESHUFFLE`
Local verification:
- `git diff --check`
- `python -m compileall
sdks/python/apache_beam/runners/portability/portable_runner.py
sdks/python/apache_beam/runners/portability/portable_runner_test.py`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
