tvalentyn opened a new issue, #40638: URL: https://github.com/apache/beam/issues/40638
### What happened? WatchFilePattern helps update a model in a RunInference pipeline on the fly. It monitors a location to find recently uploaded models and then has some logic to decide which model is the most recent one. The logic is flawed for several reasons: - `_GetLatestFileByTimeStamp` is a stateful DoFn. In Beam, state is partitioned per key, and right before calling _GetLatestFileByTimeStamp, we key each file name with its path: https://github.com/apache/beam/blob/1fc615fcd2b35b82a9b15bb3b260c989a7a64511/sdks/python/apache_beam/ml/inference/utils.py#L157 . Because keys are different and state is partitioned per key, we don't have an overview of the entirety of available models. Each new file path gets a fresh `TIME_STATE` initialized to `_START_TIME_STAMP`, files are never compared against the latest timestamp of previously seen files, so an older file uploaded after _START_TIME_STAMP may still be emitted as the "latest" model. - if more than one model is detected in the window during the poll, each with it own name, _ConvertIterToSingleton will emit a list of elements, this is well-illustrated by the existing test https://github.com/apache/beam/blob/1fc615fcd2b35b82a9b15bb3b260c989a7a64511/sdks/python/apache_beam/ml/inference/utils_test.py#L99 . When RunInference consumes this via beam.pvalue.AsSingleton(...), it will crash since the value is not a singleton. - Checking for timestamps might not be reliable for file systems that preserve the creation time when copying a file into a location. This is not a concern on GCS, since the timestamp is defined by object upload time, but if model is read from a local folder on some runners, reverting to an older model won't create desired results if some other model has a newer timestamp than the desired model to which we want to revert the processing. ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [x] Component: Python SDK - [ ] Component: Java SDK - [ ] Component: Go SDK - [ ] Component: Typescript SDK - [ ] Component: IO connector - [ ] Component: Beam YAML - [ ] Component: Beam examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Infrastructure - [ ] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Prism Runner - [ ] Component: Twister2 Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- 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]
