class_docstr: SparkML Transformer implementation. Contains a trained model,
  with which new feature data can be transformed into labels.
class_docstr_tok: ["SparkML","Transformer","implementation",".","Contains","a","trained","model",",","with","which","new","feature","data","can","be","transformed","into","labels","."]
docstr: Private transform method of a Transformer. This serves as batch-prediction method for our purposes.
          
docstr_tok: ["Private","transform","method","of","a","Transformer",".","This","serves","as","batch-prediction","method","for","our","purposes","."]
code: def _transform(self, df):
      """Private transform method of a Transformer. This serves as batch-prediction method for our purposes."""
      output_col = self.getOutputCol()
      new_schema = copy.deepcopy(df.schema)
      rdd = df.rdd
      weights = rdd.ctx.broadcast(self.weights)
  
      def batched_prediction(
          data, inference_batch_size, features_col, predict_function
      ) -> np.ndarray:
          # Do prediction in batches instead of materializing the whole partition at once
          batch = []
          preds = []
          for row in data:
              if len(batch) < inference_batch_size:
                  batch.append(from_vector(row[features_col]))
              else:
                  batch_np = np.array(batch)
                  pred = predict_function(batch_np)
                  preds.append(pred)
                  batch = [from_vector(row[features_col])]
          if len(batch) > 0:
              batch_np = np.array(batch)
              pred = predict_function(batch_np)
              preds.append(pred)
  
          if preds:
              res = np.vstack(preds)
              return res
          else:
              return np.array([])
  
      def extract_features_and_predict(
          model_json: str,
          custom_objects: dict,
          features_col: str,
          data,
          inference_batch_size: int = None,
      ):
          model = model_from_json(model_json, custom_objects)
          model.set_weights(weights.value)
          if (
              inference_batch_size is not None
              and inference_batch_size > 0
          ):
              return batched_prediction(
                  data,
                  inference_batch_size,
                  features_col,
                  model.predict,
              )
          else:
              return model.predict(
                  np.array([from_vector(x[features_col]) for x in data])
              )
  
      predictions = rdd.mapPartitions(
          partial(
              extract_features_and_predict,
              self.get_keras_model_config(),
              self.get_custom_objects(),
              self.getFeaturesCol(),
              inference_batch_size=self.get_inference_batch_size(),
          )
      )
      if self.model_type == ModelType.REGRESSION:
          predictions = predictions.map(lambda x: tuple([float(x)]))
          output_col_field = StructField(output_col, DoubleType(), True)
      else:
          # we're doing classification and predicting class probabilities
          predictions = predictions.map(lambda x: tuple([x.tolist()]))
          output_col_field = StructField(
              output_col, ArrayType(DoubleType()), True
          )
      results_rdd = rdd.zip(predictions).map(lambda x: x[0] + x[1])
  
      new_schema.add(output_col_field)
      results_df = df.sql_ctx.createDataFrame(results_rdd, new_schema)
  
      return results_df
  
imports: import copy
  import json
  import warnings
  from functools import partial
  import h5py
  import numpy
  from pyspark import keyword_only
  from pyspark.ml import Estimator, Model
  from pyspark.ml.param.shared import HasOutputCol, HasFeaturesCol, HasLabelCol
  from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable
  from pyspark.sql import DataFrame
  from pyspark.sql.types import DoubleType, StructField, ArrayType
  from tensorflow.keras.models import model_from_json
  from tensorflow.keras.optimizers import get
  from .mllib import from_vector
  from .spark_model import SparkModel
  from .utils.model_utils import LossModelTypeMapper, ModelType, ModelTypeEncoder, as_enum
  from .ml.adapter import df_to_simple_rdd
  from .ml.params import *
label_desc: Use image node_id 7 for calling the ElephasTransformer obj's underlying member method code with example usage: obj._transform(df) and returns: results_df, res, np, batched_prediction, model