Repository navigation
Conversation
is already global window in Write transform `WriteImpl._expand_bounded` previously applied `WindowInto(GlobalWindows())` unconditionally to all input records before writing bundles, even when the input `PCollection` was already using the default global windowing strategy. In typical batch pipelines, this added an unnecessary per-element `WindowInto` pass across the entire dataset (~22% overhead in bounded write benchmarks). Check `windowing.is_default()` at pipeline expansion time and only apply `WindowInto(GlobalWindows())` when the input `PCollection` is not already in the default global window.
|
context: when benchmarking a batch FileIO write pipeline (https://github.com/Abacn/apache-beam-sstable), this path alone is found to cost 10% of total CPU time
While #40910 is a more generic fix for global window instantiate, for FileIO write batch, we can skip this overhead altogether and save 10% CPU time |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #40911 +/- ##
==========================================
+ Coverage 79.83% 80.08% +0.25%
==========================================
Files 494 494
Lines 83575 83586 +11
==========================================
+ Hits 66722 66944 +222
+ Misses 16853 16642 -211
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @tvalentyn for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |

WriteImpl._expand_boundedpreviously appliedWindowInto(GlobalWindows())unconditionally to all input records before writing bundles, even when the inputPCollectionwas already using the default global windowing strategy. In typical batch pipelines, this added an unnecessary per-elementWindowIntopass across the entire dataset (~22% overhead in bounded write benchmarks).Check
windowing.is_default()at pipeline expansion time and only applyWindowInto(GlobalWindows())when the inputPCollectionis not already in the default global window.========
Benchmark Results (source: https://gist.github.com/Abacn/93fa3168450bf36838e70d853a77d2e3)
WriteImplBoundedPipeline1.643e-051.281e-05