java-spark中各种常用算子的写法示例

Function<String, LabeledPoint> transForm = new Function<String, LabeledPoint>() {//String是某一行的输入类型 LabeledPoint是转换后的输出类型
 @Override
 public LabeledPoint call(String row) throws Exception {//重写call方法
 String[] rowArr = row.split(",");
 int rowSize = rowArr.length;
 
 double[] doubleArr = new double[rowSize-1];
 
 //除了第一位的lable外 其余的部分解析成double 然后放到数组中
 for (int i = 1; i < rowSize; i++) {
  String each = rowArr[i];
  doubleArr[i] = Double.parseDouble(each);
 }
 
 //用刚才得到的数据 转成向量
 Vector feature = Vectors.dense(doubleArr);
 double label = Double.parseDouble(rowArr[0]);
 //构造用于分类训练的数据格式 LabelPoint
 LabeledPoint point = new LabeledPoint(label, feature);
 return point;
 }
 };

public class TreeEnsemble implements Serializable {
}

JavaRDD<LabeledPoint> rdd = oriData.toJavaRDD().map(transForm);

Function<String, Boolean> boolFilter = new Function<String, Boolean>() {//String是某一行的输入类型 Boolean是对应的输出类型 用于判断数据是否保留
 @Override
 public Boolean call(String row) throws Exception {//重写call方法
 boolean flag = row!=null;
 return flag;
 }
 };

JavaRDD<LabeledPoint> rdd = oriData.toJavaRDD().filter(boolFilter);

Function<String, Boolean> transformer = new PairFunction<LabeledPoint, Object, Object>() {//LabeledPoint是输入类型 后面的两个Object不要改动
 @Override
 public Tuple2 call(LabeledPoint row) throws Exception {//重写call方法 通常只改动输入参数 输出不要改动
 double predicton = thismodel.predict(row.features());
 double label = row.label();
 return new Tuple2(predicton, label);
 }
 });

JavaPairRDD<Object, Object> predictionsAndLabels = oriData.mapToPair(transformer);