Add custom field to Spark ML LabeldPoint

Question

How can I add some kind of custom field (user ID) to the forecast results?

        List<org.apache.spark.mllib.regression.LabeledPoint> localTesting = ... ;//
        // I want to add some identifier to each LabeledPoint

        DataFrame localTestDF = jsql.createDataFrame(jsc.parallelize(studyData.localTesting), LabeledPoint.class);
        DataFrame predictions = model.transform(localTestDF);
        Row[] collect = predictions.select("label", "probability", "prediction").collect();
        for (Row r : collect) {
            // and want to return identifier here.
            // so do I save I to database.
            int userNo = Integer.parseInt(r.get(0).toString());
            double prob = Double.parseDouble(r.get(1).toString());
            int prediction = Integer.parseInt(r.get(2).toString());
            log.debug(userNo + "," + prob + ", " + prediction);
        }

but when I used this class for localTesting instead of LabeledPoint,

class NoLabeledPoint extends LabeledPoint implements Serializable {
    private static final long serialVersionUID = -2488661810406135403L;
    int userNo;
    public NoLabeledPoint(double label, Vector features) {
        super(label, features);
    }

    public int getUserNo() {
        return userNo;
    }

    public void setUserNo(int userNo) {
        this.userNo = userNo;
    }
}

        List<NoLabeledPoint> localTesting = ... ;// set every user'no to the field userNo
        // I want to add some identifier to each LabeledPoint

        DataFrame localTestDF = jsql.createDataFrame(jsc.parallelize(studyData.localTesting), LabeledPoint.class);
        DataFrame predictions = model.transform(localTestDF);
        Row[] collect = predictions.select("userNo", "probability", "prediction").collect();
        for (Row r : collect) {
            // and want to return identifier here.
            // so do I save I to database.
            int userNo = Integer.parseInt(r.get(0).toString());
            double prob = Double.parseDouble(r.get(1).toString());
            int prediction = Integer.parseInt(r.get(2).toString());
            log.debug(userNo + "," + prob + ", " + prediction);
        }

the exception threw

org.apache.spark.sql.AnalysisException: cannot resolve 'userNo' given input columns rawPrediction, probability, features, label, prediction;
        at org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
        at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$2.applyOrElse(CheckAnalysis.scala:63)
        at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$2.applyOrElse(CheckAnalysis.scala:52)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:286)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:286)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:51)

I mean, I want to get not only the forecast data (function, label, probability ..), but also some kind of custom field that I want. e.g. userNo, user_id, etc. from the result: predictions.select ("......")

Update

resolved. one line should be fixed.

from

            DataFrame localTestDF = jsql.createDataFrame(jsc.parallelize(studyData.localTesting), LabeledPoint.class);

to

            DataFrame localTestDF = jsql.createDataFrame(jsc.parallelize(studyData.localTesting), NoLabeledPoint.class);

Answer 1

API MLlib , LabeledPoint . DataFrame , , Row , , *, - , .

Scala case

org.apache.spark.mllib.linalg.Vector; case class 

case class LabeledPointWithMeta(userNo: String, label: Double, features: Vector)

val rdd: RDD[LabeledPointWithMeta] = ???
val df = rdd.toDF

, , @BeanInfo:

import scala.beans.BeanInfo

@BeanInfo
case class LabeledPointWithMeta(...)

Spark SQL DataFrame Guide, , Java - **:

import org.apache.spark.mllib.linalg.Vector;

public static class LabeledPointWithMeta implements Serializable {
  private int userNo;
  private double label;
  private Vector vector;

  public int getUserNo() {
    return userNo;
  }

  public void setUserNo(int userNo) {
    this.userNo = userNo;
  }

  public double getLabel() {
    return label;
  }

  public void setLabel(double label) {
    this.label = label;
  }

  public Vector getVector() {
    return vector;
  }

  public void seVector(Vector vector) {
    this.vector = vector;
  }

}

:

JavaRDD<LabeledPointWithMeta> myPoints = ...;

DataFrame df = sqlContext.createDataFrame(myPoints LabeledPointWithMeta.class);

, :

DataFrame localTestDF = jsql.createDataFrame(
    jsc.parallelize(studyData.localTesting),
    NoLabeledPoint.class
); 

, MLlib, RDD, zip.

---

* , LabeledPoint

** , .