damccorm opened a new pull request, #41240:
URL: https://github.com/apache/beam/pull/41240
### Description
Optimizes `TimestampPrefixingWindowCoderImpl`,
`TimestampPrefixingOpaqueWindowCoderImpl`, and `ParamWindowedValueCoderImpl` in
`sdks/python/apache_beam/coders/coder_impl.py` and `coder_impl.pxd`:
1. **Reuse module-level `cdef TimestampCoderImpl _TIMESTAMP_CODER_IMPL`,
skip `Timestamp` allocation in
`TimestampPrefixingWindowCoderImpl.decode_from_stream`, and return constant `8`
bytes for timestamp size in `estimate_size`:**
- Previously, `TimestampPrefixingWindowCoderImpl` instantiated a fresh
`TimestampCoderImpl()` on every `encode_to_stream`, `decode_from_stream`, and
`estimate_size` call, and `decode_from_stream` decoded and allocated a
`Timestamp` object only to immediately discard its return value.
- Calling `stream.read_bigendian_int64()` directly in
`decode_from_stream`, reusing `_TIMESTAMP_CODER_IMPL` in `encode_to_stream`,
and using `8 + self._window_coder_impl.estimate_size(value, nested)` in
`estimate_size` eliminates per-window coder and `Timestamp` allocations during
Dataflow Runner v2 windowed shuffle.
2. **Return
`self._value_coder.get_estimated_size_and_observables(value.value,
nested=nested)` directly in `ParamWindowedValueCoderImpl` without intermediate
list allocation.**
### Benchmarks & Performance Impact
- **Strict Single-PR Isolated Microbenchmark (`upstream/master` vs
`upstream/master` + this PR alone):**
[`bench_20_timestamp_prefixing_window_coder`
(`c2feb961eef`)](https://github.com/damccorm/beam/blob/c2feb961eef2e7d2410825aab9537c62a95e6790/sdks/python/apache_beam/testing/benchmarks/python_sdk_perf/microbenchmarks.py#L324-L333)
(on branch
[`perf/py-benchmarks-suite`](https://github.com/damccorm/beam/tree/perf/py-benchmarks-suite))
— 400,000 `TimestampPrefixingWindowCoderImpl` encode + decode round-trips
improved from **`1,008.19 ms` -> `678.82 ms` (`1.49x` faster, `-32.7%`,
`-329.37 ms`)** in compiled Cython.
- **Per-Change Attribution (Most to Least Impact, 400,000 calls each in
strict isolation):**
1. *Most impact (`-363.75 ms`):*
`TimestampPrefixingWindowCoderImpl.decode_from_stream` calling
`stream.read_bigendian_int64()` instead of
`TimestampCoderImpl().decode_from_stream(stream, nested)` — **`697.65 ms` ->
`333.90 ms` (`2.09x` faster, `-52.1%`)**.
2. *Second (`-260.07 ms`):*
`TimestampPrefixingWindowCoderImpl.estimate_size` using `8 +
self._window_coder_impl.estimate_size(...)` instead of
`TimestampCoderImpl().estimate_size(value.max_timestamp())` — **`307.68 ms` ->
`47.61 ms` (`6.46x` faster, `-84.5%`)**.
3. *Least impact (`-25.30 ms`):* Reusing `_TIMESTAMP_CODER_IMPL` in
`encode_to_stream` — **`526.40 ms` -> `501.10 ms` (`1.05x` faster, `-4.8%`)**.
- **On-Worker Cloud Dataflow `n2-standard-4` Benchmark
(`p28_ms:02_ts_prefix_window_coder_400k`):** On a Cloud Dataflow
`n2-standard-4` worker, 400,000 `TimestampPrefixingWindowCoderImpl` round-trips
+ size estimates improved from **`1,740 ms` -> `1,243 ms` (`1.40x` faster,
`-28.6%`)**, and local `FnApiRunner` windowed shuffle (200,000 3-window
elements) improved from **`1,253.0 ms` -> `761.8 ms` (`1.64x` faster)**.
- **End-to-End Cloud Dataflow Benchmark (`core_window_row_combine` &
`s5-nexmark`):** Included 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)
(`3M` events -> `9M` sliding-windowed elements on `n2-standard-4`, **1.80x
faster stage execution: `599.6s` -> `332.9s`**) and `suite5_pubsub_nexmark`
(**1.71x faster stage execution: `403.0s` -> `235.0s`**).
### Testing
- Verified with `pytest sdks/python/apache_beam/coders/coders_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]
