damccorm opened a new pull request, #41159:
URL: https://github.com/apache/beam/pull/41159
### Description
Optimizes `PerWindowInvoker.invoke_process` and `_invoke_process_per_window`
in `sdks/python/apache_beam/runners/common.py` and `common.pxd`:
1. **Cythonize `PerWindowInvoker._invoke_process_per_window`:**
- Declares `cdef _invoke_process_per_window(self, WindowedValue
windowed_value, additional_args, additional_kwargs)` in `common.pxd` so Cython
dispatches per-window invocation at the C level instead of performing a dynamic
Python method lookup per element.
2. **Avoid per-element list/dict allocations on non-SDF `DoFn`s and use
`windowed_value.create`:**
- Moves `additional_args = []`, `additional_kwargs = {}`, and `residuals
= []` inside `if self.is_splittable:` (returning `()` directly on the non-SDF
path) and constructs exploded per-window `WindowedValue` instances via
`windowed_value.create(..., timestamp_micros, (w,), pane_info)` without
materializing `Timestamp` objects.
### Benchmarks & Performance Impact
- **Microbenchmark:** [`bench_08_per_window_invoker_fastpath`
(`c2feb961eef`)](https://github.com/damccorm/beam/blob/c2feb961eef2e7d2410825aab9537c62a95e6790/sdks/python/apache_beam/testing/benchmarks/python_sdk_perf/microbenchmarks.py#L166-L190)
(on branch
[`perf/py-benchmarks-suite`](https://github.com/damccorm/beam/tree/perf/py-benchmarks-suite))
— 200,000 `PerWindowInvoker.invoke_process` calls on a windowed `DoFn`
improved by **2.68x (`0.218s` -> `0.081s`, `-62.8%`)**.
- **End-to-End Cloud Dataflow Benchmark (`WindowInspectDoFn` in
`core_window_row_combine`):** Exercised directly by [`WindowInspectDoFn`
(`c2feb961eef`)](https://github.com/damccorm/beam/blob/c2feb961eef2e7d2410825aab9537c62a95e6790/sdks/python/apache_beam/testing/benchmarks/python_sdk_perf/dataflow_e2e_benchmarks.py#L197-L244)
across 9,000,000 windowed elements in [`run_core_window_row_combine_pipeline`
(`c2feb961eef`)](https://github.com/damccorm/beam/blob/c2feb961eef2e7d2410825aab9537c62a95e6790/sdks/python/apache_beam/testing/benchmarks/python_sdk_perf/dataflow_e2e_benchmarks.py#L274-L302),
where worker stage execution time improved by **1.80x (`-44.5%`, `599.6s` ->
`332.9s`)**.
### Testing
- Verified with `pytest sdks/python/apache_beam/runners/common_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]
