def _transform(self, df):
        outputCol = self.getOutputCol()
        labelCol = self.getLabelCol()
        new_schema = df.schema
        new_schema.add(StructField(outputCol, 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(outputCol, results_df[outputCol].cast(DoubleType()))
        results_df = results_df.withColumn(labelCol, results_df[labelCol].cast(DoubleType()))

        return results_df
Generate inputs, outputs numpy tensor
dataset/ETHPy150Open maxpumperla/elephas/elephas/ml_model.py/ElephasTransformer._transform--crossover with--dataset/ETHPy150Open NervanaSystems/neon/neon/data/ticker.py/Task.fetch_io