Module learning_orchestra_client.builder.builder

Classes

class BuilderSparkMl (cluster_ip: str)

Expand source code

class BuilderSparkMl:
    __TRAIN_FIELD = "trainDatasetName"
    __TEST_FIELD = "testDatasetName"
    __CODE_FIELD = "modelingCode"
    __CLASSIFIERS_LIST_FIELD = "classifiersList"

    def __init__(self, cluster_ip: str):
        self.__api_path = "/api/learningOrchestra/v1/builder/sparkml"
        self.__service_url = f'{cluster_ip}{self.__api_path}'
        self.__response_treat = ResponseTreat()
        self.__cluster_ip = cluster_ip
        self.__entity_reader = EntityReader(self.__service_url)
        self.__observer = Observer(self.__cluster_ip)

    def run_spark_ml_sync(self,
                          train_dataset_name: str,
                          test_dataset_name: str,
                          modeling_code: str,
                          model_classifiers: list,
                          pretty_response: bool = False) -> Union[dict, str]:
        """
        description: This method call runs several steps of a machine
        learning pipeline (transform, tune, train and evaluate, for instance)
        using a model code and several classifiers. It represents a way to run
        an entire pipeline. The caller waits until the method execution ends,
        since it is a synchronous method.

        train_dataset_name: Represent final train dataset.
        test_dataset_name: Represent final test dataset.
        modeling_code: Represent Python3 code for pyspark pre-processing model
        model_classifiers: list of initial classifiers to be used in the model
        pretty_response: if True it represents a result useful for visualization

        return: The set of predictions (URIs of them).
        """

        request_body_content = {
            self.__TRAIN_FIELD: train_dataset_name,
            self.__TEST_FIELD: test_dataset_name,
            self.__CODE_FIELD: modeling_code,
            self.__CLASSIFIERS_LIST_FIELD: model_classifiers,
        }
        response = requests.post(url=self.__service_url,
                                 json=request_body_content)

        for classifier in model_classifiers:
            self.__observer.wait(f'{test_dataset_name}{classifier}')

        return self.__response_treat.treatment(response, pretty_response)

    def run_spark_ml_async(self,
                           train_dataset_name: str,
                           test_dataset_name: str,
                           modeling_code: str,
                           model_classifiers: list,
                           pretty_response: bool = False) -> Union[dict, str]:
        """
        description: This method call runs several steps of a machine
        learning pipeline (transform, tune, train and evaluate, for instance)
        using a model code and several classifiers. It represents a way to run
        an entire pipeline. The caller does not wait until the method execution
        ends, since it is an asynchronous method.

        train_dataset_name: Represent final train dataset.
        test_dataset_name: Represent final test dataset.
        modeling_code: Represent Python3 code for pyspark pre-processing model
        model_classifiers: list of initial classifiers to be used in the model
        pretty_response: if True it represents a result useful for visualization

        return: the URL to retrieve the Spark pipeline result
        """

        request_body_content = {
            self.__TRAIN_FIELD: train_dataset_name,
            self.__TEST_FIELD: test_dataset_name,
            self.__CODE_FIELD: modeling_code,
            self.__CLASSIFIERS_LIST_FIELD: model_classifiers,
        }
        response = requests.post(url=self.__service_url,
                                 json=request_body_content)

        return self.__response_treat.treatment(response, pretty_response)

    def search_all_builders(self, pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description: This method retrieves all model predictions metadata. It
        does not retrieve the model predictions content.

        pretty_response: If true it returns a string, otherwise a dictionary.

        return: A list with all model predictions metadata stored in Learning
        Orchestra or an empty result.
        """

        response = self.__entity_reader.read_all_instances_from_entity()

        return self.__response_treat.treatment(response, pretty_response)

    def search_builder_register_predictions(self,
                                            builder_name: str,
                                            query: dict = {},
                                            limit: int = 10,
                                            skip: int = 0,
                                            pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description: This method is responsible for retrieving the model
        predictions content.

        pretty_response: If true it returns a string, otherwise a dictionary.
        builder_name: Represents the model predictions name.
        query: Query to make in MongoDB(default: empty query)
        limit: Number of rows to return in pagination(default: 10) (maximum is
        set at 20 rows per request)
        skip: Number of rows to skip in pagination(default: 0)

        return: A page with some tuples or registers inside or an error if the
        pipeline runs incorrectly. The current page is also returned to be used
        in future content requests.
        """

        response = self.__entity_reader.read_entity_content(
            builder_name, query, limit, skip)

        return self.__response_treat.treatment(response, pretty_response)

    def search_builder(self, builder_name: str, pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description:  This method is responsible for retrieving a specific
        model metadata.

        pretty_response: If true return indented string, else return dict.
        builder_name: Represents the model predictions name.
        limit: Number of rows to return in pagination(default: 10) (maximum is
        set at 20 rows per request)
        skip: Number of rows to skip in pagination(default: 0)

        return: Specific model prediction metadata stored in Learning Orchestra
        or an error if there is no such projections.
        """
        response = self.search_builder_register_predictions(
            builder_name, limit=1,
            pretty_response=pretty_response)

        return response

    def delete_builder(self, builder_name: str, pretty_response: bool = False) \
            -> Union[dict, str]:
        """
        description: This method is responsible for deleting a model prediction.
        The delete operation is always asynchronous,
        since the deletion is performed in background.

        pretty_response: If true it returns a string, otherwise a dictionary.
        builder_name: Represents the pipeline name.

        return: JSON object with an error message, a warning message or a
        correct delete message
        """

        cluster_url_dataset = f'{self.__service_url}/{builder_name}'

        response = requests.delete(cluster_url_dataset)

        return self.__response_treat.treatment(response, pretty_response)

    def wait(self, dataset_name: str, timeout: int = None) -> dict:
        """
           description: This method is responsible to create a synchronization
           barrier for the run_spark_ml_async method.

           dataset_name: Represents the pipeline name.
           timeout: Represents the time in seconds to wait for a builder to
           finish its run.

           return: JSON object with an error message, a warning message or a
           correct execution of a pipeline
        """
        return self.__observer.wait(dataset_name, timeout)

Methods

def delete_builder(self, builder_name: str, pretty_response: bool = False) ‑> Union[dict, str]

description: This method is responsible for deleting a model prediction. The delete operation is always asynchronous, since the deletion is performed in background.

pretty_response: If true it returns a string, otherwise a dictionary. builder_name: Represents the pipeline name.

return: JSON object with an error message, a warning message or a correct delete message

Expand source code

def delete_builder(self, builder_name: str, pretty_response: bool = False) \
        -> Union[dict, str]:
    """
    description: This method is responsible for deleting a model prediction.
    The delete operation is always asynchronous,
    since the deletion is performed in background.

    pretty_response: If true it returns a string, otherwise a dictionary.
    builder_name: Represents the pipeline name.

    return: JSON object with an error message, a warning message or a
    correct delete message
    """

    cluster_url_dataset = f'{self.__service_url}/{builder_name}'

    response = requests.delete(cluster_url_dataset)

    return self.__response_treat.treatment(response, pretty_response)

def run_spark_ml_async(self, train_dataset_name: str, test_dataset_name: str, modeling_code: str, model_classifiers: list, pretty_response: bool = False) ‑> Union[dict, str]

description: This method call runs several steps of a machine learning pipeline (transform, tune, train and evaluate, for instance) using a model code and several classifiers. It represents a way to run an entire pipeline. The caller does not wait until the method execution ends, since it is an asynchronous method.

train_dataset_name: Represent final train dataset. test_dataset_name: Represent final test dataset. modeling_code: Represent Python3 code for pyspark pre-processing model model_classifiers: list of initial classifiers to be used in the model pretty_response: if True it represents a result useful for visualization

return: the URL to retrieve the Spark pipeline result

Expand source code

def run_spark_ml_async(self,
                       train_dataset_name: str,
                       test_dataset_name: str,
                       modeling_code: str,
                       model_classifiers: list,
                       pretty_response: bool = False) -> Union[dict, str]:
    """
    description: This method call runs several steps of a machine
    learning pipeline (transform, tune, train and evaluate, for instance)
    using a model code and several classifiers. It represents a way to run
    an entire pipeline. The caller does not wait until the method execution
    ends, since it is an asynchronous method.

    train_dataset_name: Represent final train dataset.
    test_dataset_name: Represent final test dataset.
    modeling_code: Represent Python3 code for pyspark pre-processing model
    model_classifiers: list of initial classifiers to be used in the model
    pretty_response: if True it represents a result useful for visualization

    return: the URL to retrieve the Spark pipeline result
    """

    request_body_content = {
        self.__TRAIN_FIELD: train_dataset_name,
        self.__TEST_FIELD: test_dataset_name,
        self.__CODE_FIELD: modeling_code,
        self.__CLASSIFIERS_LIST_FIELD: model_classifiers,
    }
    response = requests.post(url=self.__service_url,
                             json=request_body_content)

    return self.__response_treat.treatment(response, pretty_response)

def run_spark_ml_sync(self, train_dataset_name: str, test_dataset_name: str, modeling_code: str, model_classifiers: list, pretty_response: bool = False) ‑> Union[dict, str]

description: This method call runs several steps of a machine learning pipeline (transform, tune, train and evaluate, for instance) using a model code and several classifiers. It represents a way to run an entire pipeline. The caller waits until the method execution ends, since it is a synchronous method.

train_dataset_name: Represent final train dataset. test_dataset_name: Represent final test dataset. modeling_code: Represent Python3 code for pyspark pre-processing model model_classifiers: list of initial classifiers to be used in the model pretty_response: if True it represents a result useful for visualization

return: The set of predictions (URIs of them).

Expand source code

def run_spark_ml_sync(self,
                      train_dataset_name: str,
                      test_dataset_name: str,
                      modeling_code: str,
                      model_classifiers: list,
                      pretty_response: bool = False) -> Union[dict, str]:
    """
    description: This method call runs several steps of a machine
    learning pipeline (transform, tune, train and evaluate, for instance)
    using a model code and several classifiers. It represents a way to run
    an entire pipeline. The caller waits until the method execution ends,
    since it is a synchronous method.

    train_dataset_name: Represent final train dataset.
    test_dataset_name: Represent final test dataset.
    modeling_code: Represent Python3 code for pyspark pre-processing model
    model_classifiers: list of initial classifiers to be used in the model
    pretty_response: if True it represents a result useful for visualization

    return: The set of predictions (URIs of them).
    """

    request_body_content = {
        self.__TRAIN_FIELD: train_dataset_name,
        self.__TEST_FIELD: test_dataset_name,
        self.__CODE_FIELD: modeling_code,
        self.__CLASSIFIERS_LIST_FIELD: model_classifiers,
    }
    response = requests.post(url=self.__service_url,
                             json=request_body_content)

    for classifier in model_classifiers:
        self.__observer.wait(f'{test_dataset_name}{classifier}')

    return self.__response_treat.treatment(response, pretty_response)

def search_all_builders(self, pretty_response: bool = False) ‑> Union[dict, str]

description: This method retrieves all model predictions metadata. It does not retrieve the model predictions content.

pretty_response: If true it returns a string, otherwise a dictionary.

return: A list with all model predictions metadata stored in Learning Orchestra or an empty result.

Expand source code

def search_all_builders(self, pretty_response: bool = False) \
        -> Union[dict, str]:
    """
    description: This method retrieves all model predictions metadata. It
    does not retrieve the model predictions content.

    pretty_response: If true it returns a string, otherwise a dictionary.

    return: A list with all model predictions metadata stored in Learning
    Orchestra or an empty result.
    """

    response = self.__entity_reader.read_all_instances_from_entity()

    return self.__response_treat.treatment(response, pretty_response)

def search_builder(self, builder_name: str, pretty_response: bool = False) ‑> Union[dict, str]

description: This method is responsible for retrieving a specific model metadata.

pretty_response: If true return indented string, else return dict. builder_name: Represents the model predictions name. limit: Number of rows to return in pagination(default: 10) (maximum is set at 20 rows per request) skip: Number of rows to skip in pagination(default: 0)

return: Specific model prediction metadata stored in Learning Orchestra or an error if there is no such projections.

Expand source code

def search_builder(self, builder_name: str, pretty_response: bool = False) \
        -> Union[dict, str]:
    """
    description:  This method is responsible for retrieving a specific
    model metadata.

    pretty_response: If true return indented string, else return dict.
    builder_name: Represents the model predictions name.
    limit: Number of rows to return in pagination(default: 10) (maximum is
    set at 20 rows per request)
    skip: Number of rows to skip in pagination(default: 0)

    return: Specific model prediction metadata stored in Learning Orchestra
    or an error if there is no such projections.
    """
    response = self.search_builder_register_predictions(
        builder_name, limit=1,
        pretty_response=pretty_response)

    return response

def search_builder_register_predictions(self, builder_name: str, query: dict = {}, limit: int = 10, skip: int = 0, pretty_response: bool = False) ‑> Union[dict, str]

description: This method is responsible for retrieving the model predictions content.

pretty_response: If true it returns a string, otherwise a dictionary. builder_name: Represents the model predictions name. query: Query to make in MongoDB(default: empty query) limit: Number of rows to return in pagination(default: 10) (maximum is set at 20 rows per request) skip: Number of rows to skip in pagination(default: 0)

return: A page with some tuples or registers inside or an error if the pipeline runs incorrectly. The current page is also returned to be used in future content requests.

Expand source code

def search_builder_register_predictions(self,
                                        builder_name: str,
                                        query: dict = {},
                                        limit: int = 10,
                                        skip: int = 0,
                                        pretty_response: bool = False) \
        -> Union[dict, str]:
    """
    description: This method is responsible for retrieving the model
    predictions content.

    pretty_response: If true it returns a string, otherwise a dictionary.
    builder_name: Represents the model predictions name.
    query: Query to make in MongoDB(default: empty query)
    limit: Number of rows to return in pagination(default: 10) (maximum is
    set at 20 rows per request)
    skip: Number of rows to skip in pagination(default: 0)

    return: A page with some tuples or registers inside or an error if the
    pipeline runs incorrectly. The current page is also returned to be used
    in future content requests.
    """

    response = self.__entity_reader.read_entity_content(
        builder_name, query, limit, skip)

    return self.__response_treat.treatment(response, pretty_response)

def wait(self, dataset_name: str, timeout: int = None) ‑> dict

description: This method is responsible to create a synchronization barrier for the run_spark_ml_async method.

dataset_name: Represents the pipeline name. timeout: Represents the time in seconds to wait for a builder to finish its run.

return: JSON object with an error message, a warning message or a correct execution of a pipeline

Expand source code

def wait(self, dataset_name: str, timeout: int = None) -> dict:
    """
       description: This method is responsible to create a synchronization
       barrier for the run_spark_ml_async method.

       dataset_name: Represents the pipeline name.
       timeout: Represents the time in seconds to wait for a builder to
       finish its run.

       return: JSON object with an error message, a warning message or a
       correct execution of a pipeline
    """
    return self.__observer.wait(dataset_name, timeout)