#NEWLINE# def _transform( self , df ) : #NEWLINE# #INDENT# '\n        Private transform method of a Transformer. This serves as batch-prediction method for our purposes.\n        ' #NEWLINE# outputCol = self . getOutputCol ( ) #NEWLINE# labelCol = self . getLabelCol ( ) #NEWLINE# new_schema = df . schema #NEWLINE# new_schema . add ( StructField ( outputCol , StringType ( ) , True ) ) #NEWLINE# rdd = df . rdd . coalesce ( 1 ) #NEWLINE# features = np . asarray ( rdd . map ( ( lambda x : from_vector ( x . features ) ) ) . collect ( ) ) #NEWLINE# model = model_from_yaml ( self . get_keras_model_config ( ) ) #NEWLINE# model . set_weights ( self . weights . value ) #NEWLINE# predictions = rdd . ctx . parallelize ( model . predict_classes ( features ) ) . coalesce ( 1 ) #NEWLINE# predictions = predictions . map ( ( lambda x : tuple ( str ( x ) ) ) ) #NEWLINE# results_rdd = rdd . zip ( predictions ) . map ( ( lambda x : ( x [ 0 ] + x [ 1 ] ) ) ) #NEWLINE# results_df = df . sql_ctx . createDataFrame ( results_rdd , new_schema ) #NEWLINE# results_df = results_df . withColumn ( outputCol , results_df [ outputCol ] . cast ( DoubleType ( ) ) ) #NEWLINE# results_df = results_df . withColumn ( rdd , results_df [ labelCol ] . cast ( DoubleType ( ) ) ) #NEWLINE# return results_df