jrmccluskey commented on code in PR #40667:
URL: https://github.com/apache/beam/pull/40667#discussion_r4326609123
##########
sdks/python/apache_beam/ml/inference/base.py:
##########
@@ -2078,6 +2078,9 @@ def __init__(
self._model = None
self._metrics_namespace = metrics_namespace
self._load_model_at_runtime = load_model_at_runtime
+ # The side input value that the currently loaded model reflects: a model
+ # id from ModelMetadata, a list[KeyModelPathMapping], or None if no side
+ # input value has been applied. Used to skip redundant model updates.
Review Comment:
This a pretty verbose comment that could be summarized by applying a type
annotation to the _side_input_path field instead. This should also make the
check later in the file a little easier to understand.
##########
sdks/python/apache_beam/ml/inference/base.py:
##########
@@ -2237,26 +2240,36 @@ def process(
side input is empty or the model has not been updated, the method
simply runs inference on the batch of data.
"""
- if not si_model_metadata:
- if (not self._model_metadata.is_valid_tag(self._cur_tag) or
- self._model is None):
+ if (not si_model_metadata or
+ isinstance(si_model_metadata, beam.pvalue.EmptySideInput)):
+ # No side input value: keep the loaded model, if any.
+ needs_update = (
+ self._model is None or # No model loaded yet.
+ not self._model_metadata.is_valid_tag(self._cur_tag)) # Invalidated.
+ if needs_update:
self.update_model(side_input_model_path=None)
return self._run_inference(batch, inference_args)
- if isinstance(si_model_metadata, beam.pvalue.EmptySideInput):
- self.update_model(side_input_model_path=None)
- elif isinstance(si_model_metadata, list) and hasattr(si_model_metadata[0],
- 'keys'):
+ if isinstance(si_model_metadata, list) and hasattr(si_model_metadata[0],
+ 'keys'):
# TODO(https://github.com/apache/beam/issues/27628): Update metrics here
- self.update_model(si_model_metadata)
- elif self._side_input_path != si_model_metadata.model_id:
- self._side_input_path = si_model_metadata.model_id
- self._metrics_collector = self.get_metrics_collector(
- prefix=si_model_metadata.model_name)
- lock = threading.Lock()
- with lock:
+ needs_update = (
+ self._model is None or # No model loaded yet.
+ self._side_input_path != si_model_metadata or # New mapping.
+ not self._model_metadata.is_valid_tag(self._cur_tag)) # Invalidated.
+ if needs_update:
+ self.update_model(si_model_metadata)
+ self._side_input_path = list(si_model_metadata)
+ else:
+ needs_update = (
+ self._model is None or # No model loaded yet.
+ self._side_input_path != si_model_metadata.model_id or # New model.
+ not self._model_metadata.is_valid_tag(self._cur_tag)) # Invalidated.
Review Comment:
The needs_update boolean evaluation is repeated multiple times with only
slight variations based on si_model_metadata, consider a refactor into a helper
(which also streamlines some of the repeated comments)
--
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]
