Repository navigation
[#40644][Python] Support max_batch_weight in BatchElements and honor element_size_fn in stateful and keyed RunInference batching - #40682
Conversation
…size_fn in stateful batching * Add max_batch_weight to BatchElements. When set, the batch size limits count elements and element_size_fn gives the weight of an element. This fixes RunInference failing with a TypeError when a ModelHandler sets max_batch_weight. * Honor element_size_fn when max_batch_duration_secs is set. This uses a separate stateful DoFn that only adds a 'weight' state and keeps the transform label, so pipelines that don't set element_size_fn are unaffected. * Never emit an empty batch when an element exceeds the limits on its own. * Warn when length_fn is set without max_batch_duration_secs, since it is ignored.
…NGES.md and pipeline-level regression tests Builds on the commits from apache#40603. Adds the apache#40644 issue number to the CHANGES.md entries and two RunInference-level regression tests: the pipeline construction repro from the issue (plain, keyed, maybe-keyed and stateful handlers) and an oversized example that reaches run_inference in a batch by itself without any empty batch.
There was a problem hiding this comment.
📋 Review Summary
This PR adds support for max_batch_weight to BatchElements across global, windowed, and stateful batching paths, honors element_size_fn on the stateful path while preserving DoFn labels and existing state names for streaming update compatibility, and fixes KeyedModelHandler / MaybeKeyedModelHandler to apply element_size_fn and length_fn to examples rather than (key, example) tuples. It also prevents empty batches from being emitted before oversized elements and adds comprehensive unit and integration tests.
🔍 General Feedback
- The separation of
_StatefulWeightedBatchElementsDoFnwith an overriddendefault_label()returning'_StatefulBatchElementsDoFn'is a clean design that preserves transform hierarchy and state specs for existing pipelines that do not setelement_size_fn. - Guarding batch flushes with
if self._batch:/if batch.elements:/if count:cleanly prevents empty batches when an individual element exceeds the target size or weight limits. - The
_ApplyToExamplewrapper inml/inference/base.pyis picklable and properly adapts bothelement_size_fnandlength_fnwithout mutating the underlying unkeyed handler's kwargs.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40682 +/- ##
=============================================
+ Coverage 56.16% 80.04% +23.87%
=============================================
Files 1124 490 -634
Lines 178336 82916 -95420
Branches 1489 0 -1489
=============================================
- Hits 100170 66373 -33797
+ Misses 75645 16543 -59102
+ Partials 2521 0 -2521
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: @damccorm for label python. This pull request likely touches a core component ("core" label). Please review with scrutiny. 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). |
|
Thanks, @Eliaaazzz ! |
Finishes the
max_batch_weightimplementation for RunInference, starting from the commits in #40603 (thanks, @tvalentyn; they are kept as-is with their authorship), plus the issue number inCHANGES.mdand two RunInference-level regression tests.Addresses the problems reported in #40644:
BatchElementsnow acceptsmax_batch_weight. When it is set,min_batch_size/max_batch_sizecount elements andelement_size_fngives the weight of an element; a batch is flushed before it would exceed either limit. Settingmax_batch_weighton aModelHandlerno longer fails at pipeline construction withTypeError: BatchElements.__init__() got an unexpected keyword argument 'max_batch_weight'.max_batch_duration_secs) now honorselement_size_fnandmax_batch_weight. A separate DoFn adds only aweightstate and keeps the_StatefulBatchElementsDoFnlabel, so pipelines that do not setelement_size_fnkeep using the existing DoFn and transform names stay the same (covered bytest_stateful_element_size_fn_keeps_state_and_label).run_inferencenever receives[].KeyedModelHandlerandMaybeKeyedModelHandlerapply the wrapped handler'selement_size_fnandbatch_length_fnto the example instead of the(key, example)pair.length_fnwithoutmax_batch_duration_secslogs a warning instead of being silently ignored.Argument validation at the
ModelHandlerlevel (requiring the sizing functions when batch limits are set, rejecting mixed weight-based and length-based options) is being handled separately by @alisha-1000, as discussed in #40644.BatchElementsitself raises aValueErrorwhenmax_batch_weightis set withoutelement_size_fnor is smaller than 1.Tests:
transforms/util_test.py: weight limits on the non-stateful, windowed and stateful paths, oversized elements, thelength_fnwarning, and the state/label stability check.ml/inference/base_test.py: end-to-end RunInference withmax_batch_weight(with and withoutmax_batch_duration_secs), keyed and maybe-keyed handlers sizing and bucketing examples, the pipeline construction repro from the issue, and an oversized example reachingrun_inferencealone.pytest apache_beam/transforms/util_test.py -k "Batch or Stateful"andpytest apache_beam/ml/inference/base_test.py -k "WeightedBatching or LengthAware or Keyed or batch"locally;yapfis clean.Part of #40644.
R: @tvalentyn