7adb6d43387cd054d10dfafe0b3a6f00ce2e0e4e,elephas/ml_model.py,ElephasEstimator,_fit,#ElephasEstimator#Any#,47

Before Change

        simple_rdd = df_to_simple_rdd(df, categorical=self.categorical,
                                      nb_classes=self.nb_classes)
        simple_rdd = simple_rdd.repartition(self.num_workers)
        self.train(simple_rdd, nb_epoch=self.nb_epoch,
                   batch_size=self.batch_size, verbose=self.verbose,
                   validation_split=self.validation_split)
        return ElephasTransformer(self.spark_context, self.master_network)


class ElephasTransformer(SparkModel, Transformer):

After Change

        simple_rdd = simple_rdd.repartition(self.get_num_workers())
        optimizer = None
        if self.get_optimizer_config() is not None:
            optimizer = get(self.get_optimizer_config()["name"], self.get_optimizer_config())

        spark_model = SparkModel(simple_rdd.ctx, model_from_yaml(self.get_keras_model_config()), optimizer=optimizer,
                                 mode=self.get_mode(), frequency=self.get_frequency(),
                                 num_workers=self.get_num_workers())
        spark_model.train(simple_rdd)

        model_weights = spark_model.master_network.get_weights()
        weights = simple_rdd.ctx.broadcast(model_weights)
        return ElephasTransformer(inputCol=self.getFeaturesCol(),
                                  keras_model_config=spark_model.master_network.to_yaml(),
                                  weights=weights)

Instances