RayResourceManager and ParallelLocalFoldFittingStrategy #3054
Conversation
|
Job PR-3054-eb513d4 is done. |
|
Job PR-3054-d0cf864 is done. |
| @@ -1,4 +1,5 @@ | |||
| import multiprocessing | |||
| import logging | |||
There was a problem hiding this comment.
Can we move try_import logic to common to avoid the circular dependency?
There was a problem hiding this comment.
Yea, i can do this in a follow-up PR
| class RayResourceManager: | ||
| """Manager that fetches ray cluster resources info. This class should only be used within a ray cluster.""" | ||
|
|
||
| @staticmethod | ||
| def _init_ray(): | ||
| import ray | ||
| if not ray.is_initialized(): | ||
| ray.init( | ||
| address="auto", # Force ray to connect to an existing cluster. There should be one. Otherwise, something went wrong | ||
| log_to_driver=False, | ||
| logging_level=logging.ERROR | ||
| ) | ||
|
|
||
| @staticmethod | ||
| def _get_cluster_sources(key, default_val=0): | ||
| import ray | ||
| RayResourceManager._init_ray() | ||
| return ray.cluster_resources().get(key, default_val) | ||
|
|
||
| @staticmethod | ||
| def get_cpu_count(): | ||
| return RayResourceManager._get_cluster_sources("CPU") | ||
|
|
||
| @staticmethod | ||
| def get_gpu_count_all(): | ||
| return RayResourceManager._get_cluster_sources("GPU") |
There was a problem hiding this comment.
Do we want to add some kind of sanity check assertion method such as assert_cluster_exists() that is called at the appropriate time?
There was a problem hiding this comment.
This is already done by the ray.init(address="auto"). This line will fail if there's no existing cluster
| RayResourceManager._init_ray() | ||
| return ray.cluster_resources().get(key, default_val) | ||
|
|
||
| @staticmethod |
There was a problem hiding this comment.
add type hints + consider allowing to return physical vs virtual cores.
There was a problem hiding this comment.
Added type hints. Ray cluster only work with virtual cores
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
| def _sync_model_artifact(self, local_path, model_sync_path): | ||
| pass |
There was a problem hiding this comment.
docstring + type hint. Do we even want to allow calling this in the scenarios it doesn't do anything?
There was a problem hiding this comment.
This is mainly serving as a general interface so that subclass can provide its own implementation. In the case of no syncing is needed, I think it makes sense to just leave an empty implementation there.
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
|
Job PR-3054-fa3b2f8 is done. |
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
core/src/autogluon/core/models/ensemble/fold_fitting_strategy.py
Outdated
Show resolved
Hide resolved
|
Job PR-3054-aeb30a2 is done. |
|
Job PR-3054-a17a90f is done. |
|
Job PR-3054-623a9c9 is done. |
|
Job PR-3054-873dfab is done. |
|
Job PR-3054-2600568 is done. |
Issue #, if available:
Description of changes:
RayResourceManagerandParallelLocalFoldFittingStrategyRayResourceManagerwill report total resources of the cluster instead of a single hardware.try_importfunction cannot be used here because it's being defined withinautogluon.core, which has a dependency onautogluon.common. Using it would result in a circular dependency. Here, we only instantiateRayResourceManagerwhen a env varAG_DISTRIBUTED_MODEis set. This variable is supposed to be set by thecloudmoduleParallelLocalFoldFittingStrategywill have a different ray init args and requires a s3 bucket for model synchronization between head and worker nodes. Currently the s3 bucket is being defined in a env varAG_MODEL_SYNC_PATH. This interface is subject to change.Attach a screenshot of AG being run on a cluster of 2 m5.2xlarge machine.
Take notice on how 2 node_id are present for tasks and each task is utilizing 2 cpus while a single m5.2xlarge only has 8 cpus.
predictor.predictalso works after thefitmeaning model artifacts have been saved correctlyFitting 1 L1 models ... Fitting model: LightGBM_BAG_L1 ... parallel_distributed Fitting 8 child models (S1F1 - S1F8) | Fitting with ParallelDistributedFoldFittingStrategy 0.85 = Validation score (accuracy) 0.45s = Training runtime 0.04s = Validation runtime Fitting model: WeightedEnsemble_L2 ... 0.85 = Validation score (accuracy) 0.0s = Training runtime 0.0s = Validation runtime AutoGluon training complete, total runtime = 3.73s ... Best model: "WeightedEnsemble_L2" TabularPredictor saved. To load, use: predictor = TabularPredictor.load("AutogluonModels/ag-20230316_203258/") 6118 <=50K 23204 <=50K 29590 <=50K 18116 <=50K 33964 >50K ... 29128 <=50K 23950 <=50K 13700 >50K 35248 <=50K 24772 <=50KBy submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.