feat(job): support distributed job in pipeline

client/paddleflow/job/job_info.py

@@ -13,8 +13,9 @@
 limitations under the License.
 """
 
-#!/usr/bin/env python3
+# !/usr/bin/env python3
 # -*- coding:utf8 -*-
+from paddleflow.common.exception import PaddleFlowSDKException
 
 
 class JobInfo(object):
@@ -134,8 +135,10 @@ class Member(object):
     Members
     """
 
-    def __init__(self, role, replicas, job_id=None, job_name=None, queue=None, labels=None, annotations=None, priority=None,
-                 flavour=None, fs=None, extra_fs_list=None, image=None, env=None, command=None, args_list=None, port=None,
+    def __init__(self, role, replicas, job_id=None, job_name=None, queue=None, labels=None, annotations=None,
+                 priority=None,
+                 flavour=None, fs=None, extra_fs_list=None, image=None, env=None, command=None, args_list=None,
+                 port=None,
                  extension_template=None):
         """
 
@@ -175,11 +178,55 @@ def __init__(self, role, replicas, job_id=None, job_name=None, queue=None, label
         self.port = port
         self.extension_template = extension_template
 
+    def compile(self):
+
+        result = {}
+
+        if not self.role:
+            raise PaddleFlowSDKException(f"the role of distributed member cannot be empty")
+
+        result["role"] = self.role
+
+        if not self.replicas:
+            raise PaddleFlowSDKException(f"the replicas of distributed member cannot be empty")
+
+        result["replicas"] = self.replicas
+
+        if self.image:
+            result["image"] = self.image
+
+        if self.command:
+            result["command"] = self.command
+
+        if self.flavour:
+            result["flavour"] = self.flavour
+
+        if self.fs:
+            result["fs"] = self.fs
+
+        if self.extra_fs_list:
+            result["extra_fs"] = self.extra_fs_list
+
+        if self.port:
+            result["port"] = self.port
+
+        if self.env:
+            result["env"] = self.env
+
+        if self.annotations:
+            result["annotations"] = self.annotations
+
+        if self.labels:
+            result["labels"] = self.labels
+
+        return result
+
 
 class Flavour(object):
     """
     Flavour
     """
+
     def __init__(self, name, cpu, memory, scalar_resources):
         """
 
@@ -198,12 +245,9 @@ class FileSystem(object):
     """
     FileSystem
     """
+
     def __init__(self, name, mount_path, sub_path, read_only):
         self.name = name
         self.mount_path = mount_path
         self.sub_path = sub_path
         self.read_only = read_only
-
-
-
-

client/paddleflow/pipeline/__init__.py

@@ -28,6 +28,7 @@
 from .dsl import Artifact
 from .dsl import Parameter
 from .dsl import ContainerStep
+from .dsl import DistributedJob
 from .dsl import DAG
 from .dsl import Pipeline
 from .dsl import FAIL_CONTINUE

client/paddleflow/pipeline/dsl/__init__.py

@@ -22,7 +22,7 @@
 from .options import FSOptions
 from .options import ExtraFS
 from .options import MainFS
-
+from .options import DistributedJob
 
 # import from io_types
 from .io_types import Artifact

client/paddleflow/pipeline/dsl/compiler/step_compiler.py

@@ -16,6 +16,7 @@
 from .component_compiler import ComponentCompiler
 
 from paddleflow.pipeline.dsl.options import ExtraFS
+from paddleflow.pipeline.dsl.options import DistributedJob
 from paddleflow.pipeline.dsl.utils.consts import PipelineDSLError
 from paddleflow.common.exception.paddleflow_sdk_exception import PaddleFlowSDKException
 
@@ -28,7 +29,7 @@ def __init__(self, component):
         super().__init__(component)
 
     def compile(self):
-        """ trans step to dicts'
+        """ trans step to dicts
         """
         super().compile()
 
@@ -39,9 +40,12 @@ def compile(self):
 
         # compile extra_fs
         self._compile_extra_fs()
+
+        # compile distributed_job
+        self._compile_distributed_job()
 
         return self._dict
-   
+
     def _compile_base_info(self):
         """ compiler base info such as command, docker_env, env
         """
@@ -70,7 +74,7 @@ def _compile_cache_options(self):
         """
         if self._component.cache_options:
             self._dict["cache"] = self._component.cache_options.compile()
-        
+
     def _compile_extra_fs(self):
         """ compile extra_fs 
         """
@@ -85,4 +89,10 @@ def _compile_extra_fs(self):
                 raise PaddleFlowSDKException(PipelineDSLError, 
                     self._generate_error_msg("Step's extra_fs attribute should be a list of ExtraFS instance"))
 
-            self._dict["extra_fs"].append(extra.compile())
+            self._dict["extra_fs"].append(extra.compile())
+
+    def _compile_distributed_job(self):
+        """ compile distributed_job
+        """
+        if self._component.distributed_job:
+            self._dict["distributed_job"] = self._component.distributed_job.compile()

client/paddleflow/pipeline/dsl/component/steps/container_step.py

@@ -25,6 +25,7 @@
 from paddleflow.pipeline.dsl.io_types import EnvDict
 from paddleflow.pipeline.dsl.options import CacheOptions
 from paddleflow.pipeline.dsl.options import ExtraFS
+from paddleflow.pipeline.dsl.options import DistributedJob
 from paddleflow.pipeline.dsl.utils.util import validate_string_by_regex
 from paddleflow.pipeline.dsl.utils.consts import COMPONENT_NAME_REGEX
 from paddleflow.pipeline.dsl.utils.consts import PipelineDSLError 
@@ -49,6 +50,7 @@ def __init__(
             cache_options: CacheOptions=None,
             condition: str=None,
             loop_argument: Union[List, Parameter, Artifact, str]=None,
+            distributed_job: DistributedJob=None,
             extra_fs: List[ExtraFS] = None
             ):
         """ create a new instance of ContainerStep
@@ -62,7 +64,7 @@ def __init__(
             parameters (str, Any): Parameter of step, the key is the name of this parameter, and the value could be int, string, Paramter, or upstream Step's Parameter
             env (Dict[str, str]): enviroment varible for Step runtime
             cache_options (CacheOptions): the cache options of step
-
+            distributed_job (DistributedJob): the distributed jobs defined in step
         Raises:
             PaddleFlowSDKException: if some args is illegal
         """
@@ -72,6 +74,7 @@ def __init__(
                 outputs=outputs,
                 parameters=parameters,
                 cache_options=cache_options,
+                distributed_job=distributed_job,
                 condition=condition,
                 loop_argument=loop_argument,
                 extra_fs=extra_fs

client/paddleflow/pipeline/dsl/component/steps/step.py

@@ -16,33 +16,37 @@
 import copy
 from pathlib import Path
 from typing import Dict
-from typing import Any 
-from typing import List 
+from typing import Any
+from typing import List
 from typing import Union
 
 from paddleflow.pipeline.dsl.component import Component
 from paddleflow.pipeline.dsl.io_types import Artifact
 from paddleflow.pipeline.dsl.io_types import Parameter
 from paddleflow.pipeline.dsl.options import CacheOptions
 from paddleflow.pipeline.dsl.options import ExtraFS
+from paddleflow.pipeline.dsl.options import DistributedJob
 from paddleflow.pipeline.dsl.utils.consts import PipelineDSLError
 from paddleflow.common.exception.paddleflow_sdk_exception import PaddleFlowSDKException
+from paddleflow.job import Member
 
 
 class Step(Component):
     """ Step is the basic scheduling unit in pipeline
     """
+
     def __init__(
-            self, 
-            name: str, 
-            inputs: Dict[str, Artifact]=None,
-            outputs: Dict[str, Artifact]=None,
-            parameters: Dict[str, Any]=None,
-            cache_options: CacheOptions=None,
-            condition: str=None,
-            loop_argument: Union[List, Parameter, Artifact, str]=None,
-            extra_fs: List[ExtraFS] = None
-            ):
+            self,
+            name: str,
+            inputs: Dict[str, Artifact] = None,
+            outputs: Dict[str, Artifact] = None,
+            parameters: Dict[str, Any] = None,
+            cache_options: CacheOptions = None,
+            condition: str = None,
+            loop_argument: Union[List, Parameter, Artifact, str] = None,
+            extra_fs: List[ExtraFS] = None,
+            distributed_job: DistributedJob = None,
+    ):
         """ create a new instance of Step
 
         Args:
@@ -54,25 +58,29 @@ def __init__(
             condition (str): the condition. when schedule component, would be calculate condition, if the result of it is False, then the status of this component would be set "skipped"
             loop_argument (Union[List, Parameter, Artifact, str]): the loop arugment, when schedule this component, would be be traversed, and the runtime will be created once for each item in it
             extra_fs (List[ExtraFS]): the paddleflow filesystem used by Step
+            distributed_job (DistributedJob): the distributed jobs of step
         Raises:
             PaddleFlowSDKException: if some args is illegal
         """
         super().__init__(name, inputs, outputs, parameters, condition, loop_argument)
         self._type = "step"
 
         self.cache_options = cache_options
-
+        self.distributed_job = distributed_job
 
         if extra_fs and not isinstance(extra_fs, list):
             extra_fs = [extra_fs]
-        
+
         if isinstance(extra_fs, list):
             self.extra_fs = []
             for extra in extra_fs:
                 if not isinstance(extra, ExtraFS):
-                    raise PaddleFlowSDKException(PipelineDSLError, 
-                        self._generate_error_msg("Step's extra_fs attribute should be a list of ExtraFS instance"))
+                    raise PaddleFlowSDKException(PipelineDSLError,
+                                                 self._generate_error_msg(
+                                                     "Step's extra_fs attribute should be a list of ExtraFS instance"))
 
                 self.extra_fs.append(extra)
         else:
-            self.extra_fs = []
+            self.extra_fs = []
+
+

client/paddleflow/pipeline/dsl/options/__init__.py

@@ -21,4 +21,5 @@
 from .fs_options import FSOptions
 from .fs_options import ExtraFS
 from .fs_options import MainFS
+from .distributed_job import DistributedJob
 

client/paddleflow/pipeline/dsl/options/distributed_job.py

@@ -0,0 +1,61 @@
+from paddleflow.job import Member
+
+from .options import Options
+from typing import List
+
+from paddleflow.common.exception import PaddleFlowSDKException
+from paddleflow.pipeline.dsl.utils.consts import PipelineDSLError
+
+
+class DistributedJob(object):
+    """ distributed job which used by step
+    """
+
+    def __init__(
+            self,
+            framework: str,
+            members: List[Member] = None,
+    ):
+        """ create a new instance for DistributedJob
+
+        Args:
+            framework (str): the framework of distributed job
+            members List(Member): the members defined in distributed job
+        """
+        self.framework = framework
+
+        if members:
+            if not isinstance(members, list):
+                members = [members]
+
+            for member in members:
+                if not isinstance(member, Member):
+                    raise PaddleFlowSDKException(PipelineDSLError,
+                                                 "DistributedJob's members attribute should be a list of Member instance")
+
+            self.members = members
+        else:
+            self.members = None
+
+    def compile(self):
+        """ trans to dict
+        """
+        result = {}
+
+        if not self.framework:
+            raise PaddleFlowSDKException(PipelineDSLError, "DistributedJob's framework attribute cannot empty")
+
+        result["framework"] = self.framework
+        result["members"] = []
+        if self.members:
+
+            if not isinstance(self.members, list):
+                self.members = [self.members]
+
+            for member in self.members:
+                if not isinstance(member, Member):
+                    raise PaddleFlowSDKException(PipelineDSLError,
+                                                 "DistributedJob's members attribute should be a list of Member instance")
+                result["members"].append(member.compile())
+
+        return result

client/paddleflow/pipeline/dsl/options/fs_options.py

@@ -30,7 +30,7 @@ def __init__(
             mount_path: str=None,
             read_only: bool=False 
             ):
-        """ create an new instance for ExtraFS
+        """ create a new instance for ExtraFS
 
         Args:
             name (str): the name of paddleflow filesystem

client/paddleflow/pipeline/dsl/options/options.py

@@ -26,7 +26,7 @@ def compile(self):
         """
         self._validate()
 
-        result =  {}
+        result = {}
         for attr, key in self.COMPILE_ATTR_MAP.items():
             value = getattr(self, attr, None)
             if value is not None: