java - 如何在 Spark 中管理 JavaRDD 数据结构

Question

我有一个这样的数据文件:

"id"; "a"; "b"; "c"; "d"; "e"; "f"; "outcome"
1144216938;1988;0;0;1;1965.115431348724;19.07168894289186;0
1093965329;2004;1;1;1;302.2244897959184;16.762951334379906;1
....

第一列代表每个案例的 id 标签，而最后一列代表二进制结果，即 0 或 1。中间的所有变量都用于逻辑回归模型(我尚未对它们进行标准化)。我在 Spark 中读取该文件:

JavaRDD<String> data = sc.textFile(filename);
JavaRDD<LabeledPoint> parseddata = data
    .map((String line) -> {
        String[] parts = line.split(";");
        double[] points = new double[parts.length - 1];
        for (int i = 0; i < (parts.length - 1); i++) {
            points[i] = Double.valueOf(parts[i]);
        }
        return new LabeledPoint(Double.valueOf(parts[parts.length - 1]), Vectors.dense(points));
    });

我不熟悉 LabeledPoint 数据结构，所以这可能很容易解决，但我不明白如何......所以我有三个相关问题:

• 如何去掉文件头？现在我只是将其从文件中删除:

编辑根据 Eliasah 的评论，这应该可以解决问题:

    // retrieve the header
    String first = data.first();
    // filter it out
    JavaRDD<String> filteredData = data.filter((String s) -> {
       return !s.contains(first);
    });

结束编辑

• 有没有办法将第一列映射为行名称，以便将其从回归模型中使用的点中排除 - 除了使用:

  for (int i = 1; i < (parts.length - 1); i++) {
      points[i] = Double.valueOf(parts[i]);
  }
  

• 最后，当我使用以下方法从回归模型检索预测时，如何将预测映射到正确的“id”:

  JavaRDD<Tuple2<Object, Object>> predictionAndLabels = test.map((LabeledPoint p) -> {
     Double prediction = model.predict(p.features());
     return new Tuple2<Object, Object>(prediction, p.label());
  });
  

编辑2尽管答案说明了一切，但我能想到的是:

JavaRDD<Tuple2<Object, Object>> results = filteredData.map((String line) -> {
        String[] parts = line.split(";");
        double[] points = new double[parts.length - 1];
        //skip first column with npi values
        for (int i = 1; i < (parts.length - 1); i++) {
            points[i] = Double.valueOf(parts[i]);
        }
        LabeledPoint labeledPoint = new LabeledPoint(Double.valueOf(parts[parts.length - 1]), Vectors.dense(points));
        return new Tuple2<Object, Object>(parts[0], model.predict(labeledPoint.features()));
    });

这很遗憾，因为我基本上又重新制作了 map ......这是正确的方法吗？

编辑3

感谢 eliasah 的其他评论，我现在过滤数据如下:

JavaRDD<Tuple2<String, LabeledPoint>> parsedData = filteredData.map((String line) -> {
 String[] parts = line.split(";");
 double[] points = new double[parts.length - 1];
 for (int i = 1; i < (parts.length - 1); i++) {
     points[i] = Double.valueOf(parts[i]);
 }
 return new Tuple2<String, LabeledPoint>(parts[0], new LabeledPoint(Double.valueOf(parts[parts.length - 1]), Vectors.dense(points)));
 });

我用它来训练:

JavaRDD<LabeledPoint> training = parsedData
     .map((Tuple2<String, LabeledPoint> z) -> {
         return z._2();
     });

并检索我的每个案例的概率:

JavaRDD<Tuple2<String, Double>> results = parsedData
     .map((Tuple2<String, LabeledPoint> z) -> {
         return new Tuple2<String, Double>(z._1(), model.predict(z._2().features()));
     });

Answer 1

关于 header ，我通常使用 header 是 rdd.first() 来获取 header ，然后在 rdd 上使用过滤器:

String header = data.first();
JavaRDD<String> filteredData = data.filter((String s) -> {
   return !s.contains(header);
});

关于最后一个问题，我使用 Scala 映射第二部分，然后根据 vector 是稀疏还是密集，我按如下方式创建 vector 。

你需要跟踪你的idlabeledpoint RDD[(Int,LabeledPoint)](对不起，我更熟悉scala)然后你可以通过这个rdd进行mapreduce来预测labeledpoint.feature

rdd.map{ case (id,p) => (id,model.predict(p.features)) }