gistfile1.txt

class JPMMLEvaluationMapOperator(pmmlSource: String,
  inputPreparationErrorHandlingStrategy: InputPreparationErrorHandlingStrategy,

  missingValueStrategy: MissingValueStrategy,
  resultExtractionStrategy: ResultExtractionStrategy)
    extends RichMapFunction[Map[String, Any], Map[String, Serializable]] {

  //The evaluator is transient and built in the open() function to allow proper serializability of the operator.
  @transient
  private var evaluator: Evaluator = null

  override def open(params: Configuration) = {
    super.open(params)
    evaluator = ModelEvaluatorFactory.newInstance()
      .newModelManager(
        JAXBUtil.unmarshalPMML(
          new StreamSource(
            new StringReader(pmmlSource))))
  }

  override def map(input: Map[String, Any]) = {

    val preparedInput = prepareInput(input, evaluator, missingValueStrategy, inputPreparationErrorHandlingStrategy)
    val evaluationResult = evaluate(preparedInput, evaluator)
    val result = resultExtractionStrategy(evaluationResult, evaluator)
    result
  }

}