private List<MLModelField> doGetOutputFieldsForPMMLStream(String pmmlContents) throws SAXException, JAXBException {
    List<MLModelField> fieldNames = new ArrayList<>();
    PMMLManager pmmlManager = new PMMLManager(IOUtil.unmarshal(new ByteArrayInputStream(pmmlContents.getBytes())));
    Evaluator modelEvaluator = (ModelEvaluator<?>) pmmlManager.getModelManager(null, ModelEvaluatorFactory.getInstance());
    modelEvaluator.getPredictedFields().forEach((f) -> fieldNames.add(getModelField(modelEvaluator.getDataField(f))));

    modelEvaluator.getOutputFields().forEach((f) -> {
        OutputField outputField = modelEvaluator.getOutputField(f);
        ResultFeatureType resultFeatureType = outputField.getFeature();
        if (resultFeatureType != ResultFeatureType.PREDICTED_VALUE &&
                resultFeatureType != ResultFeatureType.PREDICTED_DISPLAY_VALUE) {
            fieldNames.add(getModelField(outputField));
        }
    });

    return fieldNames;
}
 

private List<MLModelField> doGetOutputFieldsForPMMLStream(String pmmlContents) throws SAXException, JAXBException, UnsupportedEncodingException {
    List<MLModelField> fieldNames = new ArrayList<>();
    PMMLManager pmmlManager = new PMMLManager(IOUtil.unmarshal(new ByteArrayInputStream(pmmlContents.getBytes("UTF-8"))));
    Evaluator modelEvaluator = (ModelEvaluator<?>) pmmlManager.getModelManager(null, ModelEvaluatorFactory.getInstance());
    modelEvaluator.getPredictedFields().forEach((f) -> fieldNames.add(getModelField(modelEvaluator.getDataField(f))));

    modelEvaluator.getOutputFields().forEach((f) -> {
        OutputField outputField = modelEvaluator.getOutputField(f);
        ResultFeatureType resultFeatureType = outputField.getFeature();
        if (resultFeatureType != ResultFeatureType.PREDICTED_VALUE &&
                resultFeatureType != ResultFeatureType.PREDICTED_DISPLAY_VALUE) {
            fieldNames.add(getModelField(outputField));
        }
    });

    return fieldNames;
}
 

private List<MLModelField> doGetInputFieldsFromPMMLStream(String pmmlContents) throws SAXException, JAXBException {
    final List<MLModelField> fieldNames = new ArrayList<>();
    PMMLManager pmmlManager = new PMMLManager(IOUtil.unmarshal(new ByteArrayInputStream(pmmlContents.getBytes())));
    Evaluator modelEvaluator = (ModelEvaluator<?>) pmmlManager.getModelManager(null, ModelEvaluatorFactory.getInstance());
    for (FieldName predictedField : modelEvaluator.getActiveFields()) {
        fieldNames.add(getModelField(modelEvaluator.getDataField(predictedField)));
    }
    return fieldNames;
}
 

private List<MLModelField> doGetInputFieldsFromPMMLStream(String pmmlContents) throws SAXException, JAXBException, UnsupportedEncodingException {
    final List<MLModelField> fieldNames = new ArrayList<>();
    PMMLManager pmmlManager = new PMMLManager(IOUtil.unmarshal(new ByteArrayInputStream(pmmlContents.getBytes("UTF-8"))));
    Evaluator modelEvaluator = (ModelEvaluator<?>) pmmlManager.getModelManager(null, ModelEvaluatorFactory.getInstance());
    for (FieldName predictedField: modelEvaluator.getActiveFields()) {
        fieldNames.add(getModelField(modelEvaluator.getDataField(predictedField)));
    }
    return fieldNames;
}
 

private void initialize(Schema inputSchema) throws IOException, SAXException, JAXBException {

		this.inputTupleSchema = inputSchema;

		// and, initialize aliasMap:
		if (this.aliasMap == null) {
			this.aliasMap = new HashMap<String,Integer>();
			for (String alias : this.inputTupleSchema.getAliases()) {
				this.aliasMap.put(alias,this.inputTupleSchema.getPosition(alias));		// something to cleanup
			}
		}

		// Get PMML Object
		PMML pmml = null;
		try {
			
			/*
			 * TODO: Make this more robust. Specifically, Angela Ho wanted to refernce a file in the distributed
			 * 		 cache directly.  Obviously, my code doesn't support this, because it would try to open
			 * 	     the file with the IOUtil Java object, as opposed to the hadoop.fs.Path object.
			 * 
			 * TODO: This try/catch block is a hack for:
			 * 		(1) checking if execution is being done on "back-end."  A check for back-end can be done with 
			 * 			UDFContext.getUDFContext().isFrontend() BUT this does not resolve problems with local-mode.
			 * 		(2) enables testing in local-mode without failing unit tests.
			 */
			
			// Try reading file from distributed cache.
    		pmml = IOUtil.unmarshal(new File("./"+this.modelName));
    		System.err.println("Read model from distributed cache!");
    		
		} catch (Throwable t) {
			// If not on the back-end... (and distributed cache not available) ...
			
			if (this.modelPath.toLowerCase().startsWith("s3n://") || this.modelPath.toLowerCase().startsWith("s3://")) {
				// ... read from S3.
				Path path = new Path(this.modelPath);
				FileSystem fs = path.getFileSystem(new Configuration());
				FSDataInputStream in = fs.open(path);
				pmml = IOUtil.unmarshal(in);
	    		System.err.println("Read model from s3!");

			} else {
				// ... read from local file.
				pmml = IOUtil.unmarshal(new File(this.modelPath));
	    		System.err.println("Read model from local disk!");
			}

		}

		// Initialize the pmmlManager
		PMMLManager pmmlManager = new PMMLManager(pmml);
		
		// Initialize the PMML Model Manager
		ModelManager<?> modelManager = pmmlManager.getModelManager(null, ModelEvaluatorFactory.getInstance());

		this.evaluator 		 = (Evaluator)modelManager;			// Model Evaluator
		this.activeFields 	 = evaluator.getActiveFields();		// input columns
		this.predictedFields = evaluator.getPredictedFields();	// predicted columns
		this.outputFields 	 = evaluator.getOutputFields();		// derived output columns (based on predicted columns)

	}