elephas/ml_model.py
def _transform(self, df):
        """Private transform method of a Transformer. This serves as batch-prediction method for our purposes.
        """
        output_col = self.getOutputCol()
        label_col = self.getLabelCol()
        new_schema = copy.deepcopy(df.schema)
        new_schema.add(StructField(output_col, StringType(), True))

        rdd = df.rdd.coalesce(1)
        features = np.asarray(
            rdd.map(lambda x: from_vector(x.features)).collect())
        # Note that we collect, since executing this on the rdd would require model serialization once again
        model = model_from_yaml(self.get_keras_model_config())
        model.set_weights(self.weights.value)
        predictions = rdd.ctx.parallelize(
            model.predict_classes(features)).coalesce(1)
        predictions = predictions.map(lambda x: tuple(str(x)))

        results_rdd = rdd.zip(predictions).map(lambda x: x[0] + x[1])
        # TODO: Zipping like this is very likely wrong
        # results_rdd = rdd.zip(predictions).map(lambda pair: Row(features=to_vector(pair[0].features),
        #                                        label=pair[0].label, prediction=float(pair[1])))
        results_df = df.sql_ctx.createDataFrame(results_rdd, new_schema)
        results_df = results_df.withColumn(
            output_col, results_df[output_col].cast(DoubleType()))
        results_df = results_df.withColumn(
            label_col, results_df[label_col].cast(DoubleType()))

        return results_df
Private transform method of a Transformer. This serves as batch-prediction method for our purposes.
        
Private transform method of a Transformer. This serves as batch-prediction method for our purposes.
Private transform method of a Transformer . This serves as batch - prediction method for our purposes .
def _transform(self, df):
        output_col = self.getOutputCol()
        label_col = self.getLabelCol()
        new_schema = copy.deepcopy(df.schema)
        new_schema.add(StructField(output_col, StringType(), True))
        rdd = df.rdd.coalesce(1)
        features = np.asarray(
            rdd.map(lambda x: from_vector(x.features)).collect())
        model = model_from_yaml(self.get_keras_model_config())
        model.set_weights(self.weights.value)
        predictions = rdd.ctx.parallelize(
            model.predict_classes(features)).coalesce(1)
        predictions = predictions.map(lambda x: tuple(str(x)))
        results_rdd = rdd.zip(predictions).map(lambda x: x[0] + x[1])
        results_df = df.sql_ctx.createDataFrame(results_rdd, new_schema)
        results_df = results_df.withColumn(
            output_col, results_df[output_col].cast(DoubleType()))
        results_df = results_df.withColumn(
            label_col, results_df[label_col].cast(DoubleType()))
        return results_df
def _transform ( self , df ) : output_col = self . getOutputCol ( ) label_col = self . getLabelCol ( ) new_schema = copy . deepcopy ( df . schema ) new_schema . add ( StructField ( output_col , StringType ( ) , True ) ) rdd = df . rdd . coalesce ( 1 ) features = np . asarray ( rdd . map ( lambda x : from_vector ( x . features ) ) . collect ( ) ) model = model_from_yaml ( self . get_keras_model_config ( ) ) model . set_weights ( self . weights . value ) predictions = rdd . ctx . parallelize ( model . predict_classes ( features ) ) . coalesce ( 1 ) predictions = predictions . map ( lambda x : tuple ( str ( x ) ) ) results_rdd = rdd . zip ( predictions ) . map ( lambda x : x [ 0 ] + x [ 1 ] ) results_df = df . sql_ctx . createDataFrame ( results_rdd , new_schema ) results_df = results_df . withColumn ( output_col , results_df [ output_col ] . cast ( DoubleType ( ) ) ) results_df = results_df . withColumn ( label_col , results_df [ label_col ] . cast ( DoubleType ( ) ) ) return results_df
Private transform method of a Transformer.
Private transform method of a Transformer .
"""Private transform method of a Transformer. This serves as batch-prediction method for our purposes.
        """ # Note that we collect, since executing this on the rdd would require model serialization once again # TODO: Zipping like this is very likely wrong # results_rdd = rdd.zip(predictions).map(lambda pair: Row(features=to_vector(pair[0].features), #                                        label=pair[0].label, prediction=float(pair[1])))
self
df