code: 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())
          # 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])
          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
docs: Private transform method of a Transformer. This serves as batch-prediction method for our purposes.