Pipelines can request ray worker types (#289)

* add necessary changes

* refactor a bit

* mute mypy complaint

* rename parameter

* adjust docs

* readd formatting

eogrow/core/pipeline.py

@@ -171,8 +171,11 @@ def run_execution(
         executor_class: type[EOExecutor]
 
         execution_kind = self._init_processing()
+        extra_kwargs = {}
         if execution_kind is ProcessingType.RAY:
             executor_class = RayExecutor
+            if self.config.ray_worker_type is not None:
+                extra_kwargs = {"ray_remote_kwargs": {"resources": {self.config.ray_worker_type: 0.001}}}
         else:
             executor_class = EOExecutor
             executor_run_params["workers"] = self.config.workers
@@ -194,6 +197,7 @@ def run_execution(
             filesystem=self.storage.filesystem,
             logs_filter=EOExecutionFilter(ignore_packages=self.logging_manager.config.eoexecution_ignore_packages),
             logs_handler_factory=EOExecutionHandler,
+            **extra_kwargs,  # type: ignore[arg-type]
         )
         execution_results = executor.run(**executor_run_params)
 

eogrow/core/schemas.py

@@ -47,6 +47,13 @@ class PipelineSchema(BaseSchema):
     workers: int = Field(
         1, description="Number of workers for parallel execution of workflows. Parameter does not affect ray clusters."
     )
+    ray_worker_type: Optional[str] = Field(
+        description=(
+            "Restricts execution of parallelized tasks only to `ray` worker instances of the requested type. The worker"
+            " section of the `cluster.yaml` file should specify the custom resource with a matching name and the value"
+            " set to 1."
+        ),
+    )
     use_ray: BoolOrAuto = Field(
         "auto",
         description=(