java - 按行迭代 Java RDD

Question

我想遍历字符串的 RDD 并对每个字符串“做某事”。输出应该是 double[][]。这是一个带有 for 循环的示例。我知道我需要为 Java RDD 使用(我认为)foreach 函数。但是，我不知道如何理解语法。文档不是特别有用。我没有 Java 8。

这是一个示例，说明如果我可以使用常规 for 循环，我想做什么。

public class PCA {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("PCA Example");
        SparkContext sc = new SparkContext(conf);

        RDD<String> data = sc.textFile("my/directory/my/dataset.txt", 0);

        // here is the "type" of code I would like to execute
        // 30 because I have 30 variables
        double[][] vals = new double[data.count()][30];

        double[] temp;
        for (int i = 0; i < data.count(); i++) {
            temp = splitStringtoDoubles(data[i]);
            vals[i] = temp;
        }
    }

    private static double[] splitStringtoDoubles(String s) {
        String[] splitVals = s.split("\\t");
        Double[] vals = new Double[splitVals.length];
        for (int i = 0; i < splitVals.length; i++) {
            vals[i] = Double.parseDouble(splitVals[i]);
        }
    }

}

我知道 foreach 似乎需要一个返回类型为 void 的函数。不确定如何使用它。到目前为止，这是我尝试过的(显然语法错误):

    double[][] matrix = new double[data.count()][30];
    foreach(String s : data) {
        String[] splitvals = s.split("\\t");
        double[] vals = Double.parseDouble(splitvals);
        matrix[s] = vals; 
    }

Answer 1

正如 mattinbits 在评论中所说，您需要一个 map 而不是 foreach，因为您想要返回值。 map 所做的基本上是转换数据:对于 RDD 的每一行，您执行一个操作并为每一行返回一个值。你需要的可以这样实现:

import org.apache.spark.api.java.function.Function;

...

SparkConf conf = new SparkConf().setAppName("PCA Example");
SparkContext sc = new SparkContext(conf);

JavaRDD<String> data = sc.textFile("clean-sl-mix-with-labels.txt",0).toJavaRDD();
JavaRDD<double[]> whatYouWantRdd = data.map(new Function<String, double[]>() {
    @Override
    public double[] call(String row) throws Exception {
        return splitStringtoDoubles(row);
    }

    private double[] splitStringtoDoubles(String s) {
        String[] splitVals = s.split("\\t");
        Double[] vals = new Double[splitVals.length];
        for(int i=0; i < splitVals.length; i++) {
            vals[i] = Double.parseDouble(splitVals[i]);
        }
        return vals;
    }
});

List<double[]> whatYouWant = whatYouWantRdd.collect();

为了了解 Spark 的工作原理，您可以对 RDD 执行操作或转换。例如，这里我们使用 map 函数转换我们的 RDD。您需要自己创建此函数，这次使用匿名 org.apache.spark.api.java.function.Function 强制您覆盖方法 call，其中您收到一行 RDD 并返回一个值。