FateWithShellmiaoComment/Fate/python/fate_client/pipeline/backend/pipeline.py

#
#  Copyright 2019 The FATE Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
import copy
import getpass
import json
import pickle
import time
from types import SimpleNamespace

from pipeline.backend.config import Role
from pipeline.backend.config import StatusCode
from pipeline.backend.config import VERSION
from pipeline.backend.config import PipelineConfig
from pipeline.backend._operation import OnlineCommand, ModelConvert
from pipeline.backend.task_info import TaskInfo
from pipeline.component.component_base import Component
from pipeline.component.reader import Reader
from pipeline.interface import Data
from pipeline.interface import Model
from pipeline.interface import Cache
from pipeline.utils import tools
from pipeline.utils.invoker.job_submitter import JobInvoker
from pipeline.utils.logger import LOGGER
from pipeline.runtime.entity import JobParameters


class PipeLine(object):
    def __init__(self):
        self._create_time = time.asctime(time.localtime(time.time()))
        self._initiator = None
        self._roles = {}
        self._components = {}
        self._components_input = {}
        self._train_dsl = {}
        self._predict_dsl = {}
        self._train_conf = {}
        self._predict_conf = {}
        self._upload_conf = []
        self._cur_state = None
        self._job_invoker = JobInvoker()
        self._train_job_id = None
        self._predict_job_id = None
        self._fit_status = None
        self._train_board_url = None
        self._model_info = None
        self._predict_model_info = None
        self._train_components = {}
        self._stage = "fit"
        self._data_to_feed_in_prediction = None
        self._predict_pipeline = []
        self._deploy = False
        self._system_role = PipelineConfig.SYSTEM_SETTING.get("role")
        self.online = OnlineCommand(self)
        self._load = False
        self.model_convert = ModelConvert(self)
        self._global_job_provider = None

    @LOGGER.catch(reraise=True)
    def set_initiator(self, role, party_id):
        self._initiator = SimpleNamespace(role=role, party_id=party_id)
        # for predict pipeline
        if self._predict_pipeline:
            predict_pipeline = self._predict_pipeline[0]["pipeline"]
            predict_pipeline._initiator = SimpleNamespace(role=role, party_id=party_id)

        return self

    def get_component_list(self):
        return copy.copy(list(self._components.keys()))

    def restore_roles(self, initiator, roles):
        self._initiator = initiator
        self._roles = roles

    @LOGGER.catch(reraise=True)
    def get_predict_meta(self):
        if self._fit_status != StatusCode.SUCCESS:
            raise ValueError("To get predict meta, please fit successfully")

        return {"predict_dsl": self._predict_dsl,
                "train_conf": self._train_conf,
                "initiator": self._initiator,
                "roles": self._roles,
                "model_info": self._model_info,
                "components": self._components,
                "stage": self._stage
                }

    def get_predict_model_info(self):
        return copy.deepcopy(self._predict_model_info)

    def get_model_info(self):
        return copy.deepcopy(self._model_info)

    def get_train_dsl(self):
        return copy.deepcopy(self._train_dsl)

    def get_train_conf(self):
        return copy.deepcopy(self._train_conf)

    def get_predict_dsl(self):
        return copy.deepcopy(self._predict_dsl)

    def get_predict_conf(self):
        return copy.deepcopy(self._predict_conf)

    def get_upload_conf(self):
        return copy.deepcopy(self._upload_conf)

    def _get_initiator_conf(self):
        if self._initiator is None:
            raise ValueError("Please set initiator of PipeLine")

        initiator_conf = {"role": self._initiator.role,
                          "party_id": self._initiator.party_id}

        return initiator_conf

    def set_global_job_provider(self, provider):
        self._global_job_provider = provider
        return self

    @LOGGER.catch(reraise=True)
    def set_roles(self, guest=None, host=None, arbiter=None, **kwargs):
        local_parameters = locals()
        support_roles = Role.support_roles()
        for role, party_id in local_parameters.items():
            if role == "self":
                continue

            if not local_parameters.get(role):
                continue

            if role not in support_roles:
                raise ValueError("Current role not support {}, support role list {}".format(role, support_roles))

            party_id = local_parameters.get(role)
            self._roles[role] = []
            if isinstance(party_id, int):
                self._roles[role].append(party_id)
            elif isinstance(party_id, list):
                self._roles[role].extend(party_id)
            else:
                raise ValueError("role: {}'s party_id should be an integer or a list of integer".format(role))
            # update role config for compiled pipeline
            if self._train_conf:
                if role in self._train_conf["role"]:
                    self._train_conf["role"][role] = self._roles[role]

        if self._predict_pipeline:
            predict_pipeline = self._predict_pipeline[0]["pipeline"]
            predict_pipeline._roles = self._roles

        return self

    def _get_role_conf(self):
        return self._roles

    def _get_party_index(self, role, party_id):
        if role not in self._roles:
            raise ValueError("role {} does not setting".format(role))

        if party_id not in self._roles[role]:
            raise ValueError("role {} does not init setting with the party_id {}".format(role, party_id))

        return self._roles[role].index(party_id)

    @LOGGER.catch(reraise=True)
    def add_component(self, component, data=None, model=None, cache=None):
        if isinstance(component, PipeLine):
            if component.is_deploy() is False:
                raise ValueError("To use a training pipeline object as predict component, should deploy model first")

            if model:
                raise ValueError("pipeline should not have model as input!")

            if not data:
                raise ValueError("To use pipeline as a component, please set data input")

            self._stage = "predict"
            self._predict_pipeline.append({"pipeline": component, "data": data.predict_input})

            meta = component.get_predict_meta()
            self.restore_roles(meta.get("initiator"), meta.get("roles"))

            return self

        if not isinstance(component, Component):
            raise ValueError(
                "To add a component to pipeline, component {} should be a Component object".format(component))

        if component.name in self._components:
            raise Warning("component {} is added before".format(component.name))

        self._components[component.name] = component

        if data is not None:
            if not isinstance(data, Data):
                raise ValueError("data input of component {} should be passed by data object".format(component.name))

            attrs_dict = vars(data)
            self._components_input[component.name] = {"data": {}}
            for attr, val in attrs_dict.items():
                if not attr.endswith("data"):
                    continue

                if val is None:
                    continue

                data_key = attr.strip("_")

                if isinstance(val, list):
                    self._components_input[component.name]["data"][data_key] = val
                else:
                    self._components_input[component.name]["data"][data_key] = [val]

        if model is not None:
            if not isinstance(model, Model):
                raise ValueError("model input of component {} should be passed by model object".format(component.name))

            attrs_dict = vars(model)
            for attr, val in attrs_dict.items():
                if not attr.endswith("model"):
                    continue

                if val is None:
                    continue

                if isinstance(val, list):
                    self._components_input[component.name][attr.strip("_")] = val
                else:
                    self._components_input[component.name][attr.strip("_")] = [val]

        if cache is not None:
            if not isinstance(cache, Cache):
                raise ValueError("cache input of component {} should be passed by cache object".format(component.name))

            attr = cache.cache
            if not isinstance(attr, list):
                attr = [attr]

            self._components_input[component.name]["cache"] = attr

        return self

    @LOGGER.catch(reraise=True)
    def add_upload_data(self, file, table_name, namespace, head=1, partition=16,
                        id_delimiter=",", extend_sid=False, auto_increasing_sid=False, **kargs):
        data_conf = {"file": file,
                     "table_name": table_name,
                     "namespace": namespace,
                     "head": head,
                     "partition": partition,
                     "id_delimiter": id_delimiter,
                     "extend_sid": extend_sid,
                     "auto_increasing_sid": auto_increasing_sid, **kargs}
        self._upload_conf.append(data_conf)

    def _get_task_inst(self, job_id, name, init_role, party_id):
        component = None
        if name in self._components:
            component = self._components[name]

        if component is None:
            if self._stage != "predict":
                raise ValueError(f"Component {name} does not exist")
            training_meta = self._predict_pipeline[0]["pipeline"].get_predict_meta()

            component = training_meta.get("components").get(name)

            if component is None:
                raise ValueError(f"Component {name} does not exist")

        return TaskInfo(jobid=job_id,
                        component=component,
                        job_client=self._job_invoker,
                        role=init_role,
                        party_id=party_id)

    @LOGGER.catch(reraise=True)
    def get_component(self, component_names=None):
        job_id = self._train_job_id
        if self._cur_state != "fit":
            job_id = self._predict_job_id

        init_role = self._initiator.role
        party_id = self._initiator.party_id
        if not component_names:
            component_tasks = {}
            for name in self._components:
                component_tasks[name] = self._get_task_inst(job_id, name, init_role, party_id)
            return component_tasks
        elif isinstance(component_names, str):
            return self._get_task_inst(job_id, component_names, init_role, party_id)
        elif isinstance(component_names, list):
            component_tasks = []
            for name in component_names:
                component_tasks.append(self._get_task_inst(job_id, name, init_role, party_id))

            return component_tasks

    def _construct_train_dsl(self):
        if self._global_job_provider:
            self._train_dsl["provider"] = self._global_job_provider

        self._train_dsl["components"] = {}
        for name, component in self._components.items():
            component_dsl = {"module": component.module}
            if name in self._components_input:
                component_dsl["input"] = self._components_input[name]

            if hasattr(component, "output"):
                component_dsl["output"] = {}
                output_attrs = {"data": "data_output",
                                "model": "model_output",
                                "cache": "cache_output"}

                for output_key, attr in output_attrs.items():
                    if hasattr(component.output, attr):
                        component_dsl["output"][output_key] = getattr(component.output, attr)

            provider_name = None
            provider_version = None
            if not hasattr(component, "source_provider"):
                LOGGER.warning(f"Can not retrieval source provider of component {name}, "
                               f"refer to pipeline/component/component_base.py")
            else:
                provider_name = getattr(component, "source_provider")
                if provider_name is None:
                    LOGGER.warning(f"Source provider of component {name} is None, "
                                   f"refer to pipeline/component/component_base.py")

            if hasattr(component, "provider"):
                provider = getattr(component, "provider")
                if provider is not None:
                    if provider.find("@") != -1:
                        provider_name, provider_version = provider.split("@", -1)
                    else:
                        provider_name = provider
                    # component_dsl["provider"] = provider

            if getattr(component, "provider_version") is not None:
                provider_version = getattr(component, "provider_version")

            if provider_name and provider_version:
                component_dsl["provider"] = "@".join([provider_name, provider_version])
            elif provider_name:
                component_dsl["provider"] = provider_name

            self._train_dsl["components"][name] = component_dsl

        if not self._train_dsl:
            raise ValueError("there are no components to train")

        LOGGER.debug(f"train_dsl: {self._train_dsl}")

    def _construct_train_conf(self):
        self._train_conf["dsl_version"] = VERSION
        self._train_conf["initiator"] = self._get_initiator_conf()
        self._train_conf["role"] = self._roles
        self._train_conf["job_parameters"] = {"common": {"job_type": "train"}}
        for name, component in self._components.items():
            param_conf = component.get_config(version=VERSION, roles=self._roles)
            if "common" in param_conf:
                common_param_conf = param_conf["common"]
                if "component_parameters" not in self._train_conf:
                    self._train_conf["component_parameters"] = {}
                if "common" not in self._train_conf["component_parameters"]:
                    self._train_conf["component_parameters"]["common"] = {}

                self._train_conf["component_parameters"]["common"].update(common_param_conf)

            if "role" in param_conf:
                role_param_conf = param_conf["role"]
                if "component_parameters" not in self._train_conf:
                    self._train_conf["component_parameters"] = {}
                if "role" not in self._train_conf["component_parameters"]:
                    self._train_conf["component_parameters"]["role"] = {}
                self._train_conf["component_parameters"]["role"] = tools.merge_dict(
                    role_param_conf, self._train_conf["component_parameters"]["role"])

        LOGGER.debug(f"self._train_conf: \n {json.dumps(self._train_conf, indent=4, ensure_ascii=False)}")
        return self._train_conf

    def _construct_upload_conf(self, data_conf):
        upload_conf = copy.deepcopy(data_conf)
        # upload_conf["work_mode"] = work_mode
        return upload_conf

    def describe(self):
        LOGGER.info(f"Pipeline Stage is {self._stage}")
        LOGGER.info("DSL is:")
        if self._stage == "fit":
            LOGGER.info(f"{self._train_dsl}")
        else:
            LOGGER.info(f"{self._predict_dsl}")

        LOGGER.info(f"Pipeline Create Time: {self._create_time}")

    def get_train_job_id(self):
        return self._train_job_id

    def get_predict_job_id(self):
        return self._predict_job_id

    def _set_state(self, state):
        self._cur_state = state

    def set_job_invoker(self, job_invoker):
        self._job_invoker = job_invoker

    @LOGGER.catch(reraise=True)
    def compile(self):
        self._construct_train_dsl()
        self._train_conf = self._construct_train_conf()
        if self._stage == "predict":
            predict_pipeline = self._predict_pipeline[0]["pipeline"]
            data_info = self._predict_pipeline[0]["data"]

            meta = predict_pipeline.get_predict_meta()
            if meta["stage"] == "predict":
                raise ValueError(
                    "adding predict pipeline objects'stage is predict, a predict pipeline cannot be an input component")

            self._model_info = meta["model_info"]
            predict_pipeline_dsl = meta["predict_dsl"]
            predict_pipeline_conf = meta["train_conf"]
            if not predict_pipeline_dsl:
                raise ValueError(
                    "Cannot find deploy model in predict pipeline, to use a pipeline as input component, "
                    "it should be deploy first")

            for cpn in self._train_dsl["components"]:
                if cpn in predict_pipeline_dsl["components"]:
                    raise ValueError(
                        f"component name {cpn} exist in predict pipeline's deploy component, this is not support")

            if "algorithm_parameters" in predict_pipeline_conf:
                algo_param = predict_pipeline_conf["algorithm_parameters"]
                if "algorithm_parameters" in self._train_conf:
                    for key, value in algo_param.items():
                        if key not in self._train_conf["algorithm_parameters"]:
                            self._train_conf["algorithm_parameters"][key] = value
                else:
                    self._train_conf["algorithm_parameters"] = algo_param

            if "role_parameters" in predict_pipeline_conf:
                role_param = predict_pipeline_conf["role_parameters"]
                for cpn in self._train_dsl["components"]:
                    for role, param in role_param.items():
                        for idx in param:
                            if param[idx].get(cpn) is not None:
                                del predict_pipeline_conf["role_parameters"][role][idx][cpn]

                if "role_parameters" not in self._train_conf:
                    self._train_conf["role_parameters"] = {}

                self._train_conf["role_parameters"] = tools.merge_dict(self._train_conf["role_parameters"],
                                                                       predict_pipeline_conf["role_parameters"])

            self._predict_dsl = tools.merge_dict(predict_pipeline_dsl, self._train_dsl)

            for data_field, val in data_info.items():
                cpn = data_field.split(".", -1)[0]
                dataset = data_field.split(".", -1)[1]
                if not isinstance(val, list):
                    val = [val]

                if "input" not in self._predict_dsl["components"][cpn]:
                    self._predict_dsl["components"][cpn]["input"] = {}

                if 'data' not in self._predict_dsl["components"][cpn]["input"]:
                    self._predict_dsl["components"][cpn]["input"]["data"] = {}

                self._predict_dsl["components"][cpn]["input"]["data"][dataset] = val

        return self

    @LOGGER.catch(reraise=True)
    def _check_duplicate_setting(self, submit_conf):
        system_role = self._system_role
        if "role" in submit_conf["job_parameters"]:
            role_conf = submit_conf["job_parameters"]["role"]
            system_role_conf = role_conf.get(system_role, {})
            for party, conf in system_role_conf.items():
                if conf.get("user"):
                    raise ValueError(f"system role {system_role}'s user info already set. Please check.")

    def _feed_job_parameters(self, conf, job_type=None,
                             model_info=None, job_parameters=None):
        submit_conf = copy.deepcopy(conf)
        LOGGER.debug(f"submit conf type is {type(submit_conf)}")

        if job_parameters:
            submit_conf["job_parameters"] = job_parameters.get_config(roles=self._roles)

        if "common" not in submit_conf["job_parameters"]:
            submit_conf["job_parameters"]["common"] = {}

        submit_conf["job_parameters"]["common"]["job_type"] = job_type

        if model_info is not None:
            submit_conf["job_parameters"]["common"]["model_id"] = model_info.model_id
            submit_conf["job_parameters"]["common"]["model_version"] = model_info.model_version

        if self._system_role:
            self._check_duplicate_setting(submit_conf)
            init_role = self._initiator.role
            idx = str(self._roles[init_role].index(self._initiator.party_id))
            if "role" not in submit_conf["job_parameters"]:
                submit_conf["job_parameters"]["role"] = {}

            if init_role not in submit_conf["job_parameters"]["role"]:
                submit_conf["job_parameters"]["role"][init_role] = {}

            if idx not in submit_conf["job_parameters"]["role"][init_role]:
                submit_conf["job_parameters"]["role"][init_role][idx] = {}

            submit_conf["job_parameters"]["role"][init_role][idx].update({"user": getpass.getuser()})
        return submit_conf

    def _filter_out_deploy_component(self, predict_conf):
        if "component_parameters" not in predict_conf:
            return predict_conf

        if "common" in predict_conf["component_parameters"]:
            cpns = list(predict_conf["component_parameters"]["common"])
            for cpn in cpns:
                if cpn not in self._components.keys():
                    del predict_conf["component_parameters"]["common"]

        if "role" in predict_conf["component_parameters"]:
            roles = predict_conf["component_parameters"]["role"].keys()
            for role in roles:
                role_params = predict_conf["component_parameters"]["role"].get(role)
                indexs = role_params.keys()
                for idx in indexs:
                    cpns = role_params[idx].keys()
                    for cpn in cpns:
                        if cpn not in self._components.keys():
                            del role_params[idx][cpn]

                    if not role_params[idx]:
                        del role_params[idx]

                if role_params:
                    predict_conf["component_parameters"]["role"][role] = role_params
                else:
                    del predict_conf["component_parameters"]["role"][role]

        return predict_conf

    @LOGGER.catch(reraise=True)
    def fit(self, job_parameters=None, callback_func=None):

        if self._stage == "predict":
            raise ValueError("This pipeline is constructed for predicting, cannot use fit interface")

        if job_parameters and not isinstance(job_parameters, JobParameters):
            raise ValueError("input parameter of fit function should be JobParameters object")

        LOGGER.debug(f"in fit, _train_conf is: \n {json.dumps(self._train_conf)}")
        self._set_state("fit")
        training_conf = self._feed_job_parameters(self._train_conf, job_type="train", job_parameters=job_parameters)
        self._train_conf = training_conf
        LOGGER.debug(f"train_conf is: \n {json.dumps(training_conf, indent=4, ensure_ascii=False)}")
        self._train_job_id, detail_info = self._job_invoker.submit_job(self._train_dsl, training_conf, callback_func)
        self._train_board_url = detail_info["board_url"]
        self._model_info = SimpleNamespace(model_id=detail_info["model_info"]["model_id"],
                                           model_version=detail_info["model_info"]["model_version"])

        self._fit_status = self._job_invoker.monitor_job_status(self._train_job_id,
                                                                self._initiator.role,
                                                                self._initiator.party_id)

    @LOGGER.catch(reraise=True)
    def update_model_info(self, model_id=None, model_version=None):
        # predict pipeline
        if self._predict_pipeline:
            predict_pipeline = self._predict_pipeline[0]["pipeline"]
            if model_id:
                predict_pipeline._model_info.model_id = model_id
            if model_version:
                predict_pipeline._model_info.model_version = model_version
            return self
        # train pipeline
        original_model_id, original_model_version = None, None
        if self._model_info is not None:
            original_model_id, original_model_version = self._model_info.model_id, self._model_info.model_version
        new_model_id = model_id if model_id is not None else original_model_id
        new_model_version = model_version if model_version is not None else original_model_version
        if new_model_id is None and new_model_version is None:
            return self
        self._model_info = SimpleNamespace(model_id=new_model_id, model_version=new_model_version)
        return self

    @LOGGER.catch(reraise=True)
    def continuously_fit(self):
        self._fit_status = self._job_invoker.monitor_job_status(self._train_job_id,
                                                                self._initiator.role,
                                                                self._initiator.party_id,
                                                                previous_status=self._fit_status)

    @LOGGER.catch(reraise=True)
    def predict(self, job_parameters=None, components_checkpoint=None):
        """

        Parameters
        ----------
        job_parameters: None
        components_checkpoint: specify which model to take, ex.: {"hetero_lr_0": {"step_index": 8}}

        Returns
        -------

        """
        if self._stage != "predict":
            raise ValueError(
                "To use predict function, please deploy component(s) from training pipeline"
                "and construct a new predict pipeline with data reader and training pipeline.")

        if job_parameters and not isinstance(job_parameters, JobParameters):
            raise ValueError("input parameter of fit function should be JobParameters object")

        self.compile()

        res_dict = self._job_invoker.model_deploy(model_id=self._model_info.model_id,
                                                  model_version=self._model_info.model_version,
                                                  predict_dsl=self._predict_dsl,
                                                  components_checkpoint=components_checkpoint)
        self._predict_model_info = SimpleNamespace(model_id=res_dict["model_id"],
                                                   model_version=res_dict["model_version"])
        predict_conf = self._feed_job_parameters(self._train_conf,
                                                 job_type="predict",
                                                 model_info=self._predict_model_info,
                                                 job_parameters=job_parameters)
        predict_conf = self._filter_out_deploy_component(predict_conf)
        self._predict_conf = copy.deepcopy(predict_conf)
        predict_dsl = copy.deepcopy(self._predict_dsl)

        self._predict_job_id, _ = self._job_invoker.submit_job(dsl=predict_dsl, submit_conf=predict_conf)
        self._job_invoker.monitor_job_status(self._predict_job_id,
                                             self._initiator.role,
                                             self._initiator.party_id)

    @LOGGER.catch(reraise=True)
    def upload(self, drop=0):
        for data_conf in self._upload_conf:
            upload_conf = self._construct_upload_conf(data_conf)
            LOGGER.debug(f"upload_conf is {json.dumps(upload_conf)}")
            self._train_job_id, detail_info = self._job_invoker.upload_data(upload_conf, int(drop))
            self._train_board_url = detail_info["board_url"]
            self._job_invoker.monitor_job_status(self._train_job_id,
                                                 "local",
                                                 0)

    @LOGGER.catch(reraise=True)
    def dump(self, file_path=None):
        pkl = pickle.dumps(self)

        if file_path is not None:
            with open(file_path, "wb") as fout:
                fout.write(pkl)

        return pkl

    @classmethod
    def load(cls, pipeline_bytes):
        """
        return pickle.loads(pipeline_bytes)
        """
        pipeline_obj = pickle.loads(pipeline_bytes)
        pipeline_obj.set_job_invoker(JobInvoker())
        return pipeline_obj

    @classmethod
    def load_model_from_file(cls, file_path):
        with open(file_path, "rb") as fin:
            pipeline_obj = pickle.loads(fin.read())
            pipeline_obj.set_job_invoker(JobInvoker())
            return pipeline_obj

    @LOGGER.catch(reraise=True)
    def deploy_component(self, components=None):
        if self._train_dsl is None:
            raise ValueError("Before deploy model, training should be finished!!!")

        if components is None:
            components = self._components
        deploy_cpns = []
        for cpn in components:
            if isinstance(cpn, str):
                deploy_cpns.append(cpn)
            elif isinstance(cpn, Component):
                deploy_cpns.append(cpn.name)
            else:
                raise ValueError(
                    "deploy component parameters is wrong, expect str or Component object, but {} find".format(
                        type(cpn)))

            if deploy_cpns[-1] not in self._components:
                raise ValueError("Component {} does not exist in pipeline".format(deploy_cpns[-1]))

            if isinstance(self._components.get(deploy_cpns[-1]), Reader):
                raise ValueError("Reader should not be include in predict pipeline")

        res_dict = self._job_invoker.model_deploy(model_id=self._model_info.model_id,
                                                  model_version=self._model_info.model_version,
                                                  cpn_list=deploy_cpns)
        self._predict_model_info = SimpleNamespace(model_id=res_dict["model_id"],
                                                   model_version=res_dict["model_version"])

        self._predict_dsl = self._job_invoker.get_predict_dsl(model_id=res_dict["model_id"],
                                                              model_version=res_dict["model_version"])

        if self._predict_dsl:
            self._deploy = True

        return self

    def is_deploy(self):
        return self._deploy

    def is_load(self):
        return self._load

    @LOGGER.catch(reraise=True)
    def init_predict_config(self, config):
        if isinstance(config, PipeLine):
            config = config.get_predict_meta()

        self._stage = "predict"
        self._model_info = config["model_info"]
        self._predict_dsl = config["predict_dsl"]
        self._train_conf = config["train_conf"]
        self._initiator = config["initiator"]
        self._train_components = config["train_components"]

    @LOGGER.catch(reraise=True)
    def get_component_input_msg(self):
        if VERSION != 2:
            raise ValueError("In DSL Version 1，only need to config data from args, do not need special component")

        need_input = {}
        for cpn_name, config in self._predict_dsl["components"].items():
            if "input" not in config:
                continue

            if "data" not in config["input"]:
                continue

            data_config = config["input"]["data"]
            for data_type, dataset_list in data_config.items():
                for data_set in dataset_list:
                    input_cpn = data_set.split(".", -1)[0]
                    input_inst = self._components[input_cpn]
                    if isinstance(input_inst, Reader):
                        if cpn_name not in need_input:
                            need_input[cpn_name] = {}

                        need_input[cpn_name][data_type] = []
                        need_input[cpn_name][data_type].append(input_cpn)

        return need_input

    @LOGGER.catch(reraise=True)
    def get_input_reader_placeholder(self):
        input_info = self.get_component_input_msg()
        input_placeholder = set()
        for cpn_name, data_dict in input_info.items():
            for data_type, dataset_list in data_dict.items():
                for dataset in dataset_list:
                    input_placeholder.add(dataset)

        return input_placeholder

    @LOGGER.catch(reraise=True)
    def set_inputs(self, data_dict):
        if not isinstance(data_dict, dict):
            raise ValueError(
                "inputs for predicting should be a dict, key is input_placeholder name, value is a reader object")

        unfilled_placeholder = self.get_input_reader_placeholder() - set(data_dict.keys())
        if unfilled_placeholder:
            raise ValueError("input placeholder {} should be fill".format(unfilled_placeholder))

        self._data_to_feed_in_prediction = data_dict

    @LOGGER.catch(reraise=True)
    def bind_table(self, name, namespace, path, engine='PATH', replace=True, **kwargs):
        info = self._job_invoker.bind_table(engine=engine, name=name, namespace=namespace, address={
            "path": path
        }, drop=replace, **kwargs)
        return info

    # @LOGGER.catch(reraise=True)
    def __getattr__(self, attr):
        if attr in self._components:
            return self._components[attr]

        return self.__getattribute__(attr)

    @LOGGER.catch(reraise=True)
    def __getitem__(self, item):
        if item not in self._components:
            raise ValueError("Pipeline does not has component }{}".format(item))

        return self._components[item]

    def __getstate__(self):
        return vars(self)

    def __setstate__(self, state):
        vars(self).update(state)