如何使用Stream Parallel Java进行矩阵计算？

Question

我正在尝试使用多维数组(非常大的行)创建一个矩阵算术运算方法。我是个新手，我就是找不到我做错了什么。如果您能告诉我这是什么，我将不胜感激。

    try {
        Stream<String> Matrix = Files.lines(Paths.get(file)).parallel();
        String[][] DataSet = Matrix.map(mapping -> mapping.split(",")).toArray(String[][]::new);
        Double[][] distanceTable = new Double[DataSet.length - 1][];

        /* START WANT TO REPLACE THIS MATRIX CALCULATION WITH PARALLEL STREAM RATHER THAN USE TRADITIONAL ARRAY ARITHMETICS START  */

        for (int i = 0; i < distanceTable.length - 1; ++i) {
            distanceTable[i] = new Double[i + 1];
            for (int j = 0; j <= i; ++j) {
                double distance = 0.0;
                for (int k = 0; k < DataSet[i + 1].length; ++k) {
                    double difference = Double.parseDouble(DataSet[j][k]) - Double.parseDouble(DataSet[i + 1][k]);
                    distance += difference * difference;
                }
                distanceTable[i][j] = distance;
            }
        }

        /* END WANT TO REPLACE THIS MATRIX CALCULATION WITH PARALLEL STREAM RATHER THAN USE TRADITIONAL ARRAY ARITHMETICS START  */

        } catch ( Exception except ){
            System.out.println ( except );
        }

我宁愿不使用库或类似的东西，我这样做主要是为了了解它是如何工作的。非常感谢你提前这么做。如果你询问的数据是这样的：

4,53
5,63
10,59
9,77
13,49

数据处理的输出应如下所示：

[101] <- ((4-5)^2) + ((53-63)^2)
[72, 41] <- ( ((4-10)^2) + ((53-59)^2) ), ( ((5,10)^2) + ((63-59)^2))
[601.0, 212.0, 325.0]
[97.0, 260.0, 109.0, 800.0]
[337.0, 100.0, 109.0, 80.0, 400.0]

Answer 1

我尝试用distanceTable更改matrixDistance。尝试将这段代码移到不同的方法中，这样就可以并行运行它

        for(int i = 0; i < matrixDistance.length - 1; ++i) {
            distanceTable[i] = new double[i + 1];
            for(int j = 0; j <= i; ++j) {
                double distance = 0.0;
                for(int k = 0; k < DataSet[i+1].length; ++k) {
                    double difference = Double.parseDouble(DataSet[j][k]) - Double.parseDouble(DataSet[i+1][k]);
                    distance += difference * difference;
                }
                distanceTable[i][j] = distance;
            }
        }

我已经根据你的问题创建了这个例子。

    public void parallel(String file)
    ....
    // parsing from csv into matrix 2d Double[][]
    ....
        IntStream
            .range(1, data.length - 1)
            .parallel()
            .forEach(i -> {
                add(euclidian.euclidian(Arrays.copyOf(data, i+1)), i);
            });
}

这是你的算法的迷你版。

    public Double[] euclidian(Double[][] data) {
        Double[] result = new Double[data.length - 1];
        for (int i = 0; i < result.length; i++) {
            result[i] =
                    Math.pow(data[i][0] - data[data.length - 1][0], 2) +
                            Math.pow(data[i][1] - data[data.length - 1][1], 2);
        }

        return result;
    }

而且由于是并行执行，所以需要在distanceTable中添加用于插入数据的锁定方法。

    private final Object lock = new Object();
    Double[][] distanceTable;

    void add(Double[] data, int index){
        synchronized (lock) {
            distanceTable[index - 1] = data;
        }
    }

我已经在我的笔记本电脑上对csv文件中的74行进行了测试，比较结果是这样的(ORI使用您的代码，PAR使用我的方法)：

java -jar target/stream-example-1.0-SNAPSHOT.jar test.csv 
#####################
ORI read: 59 ms
ORI  map: 71 ms
ORI time: 80 ms
#####################
PAR read: 0 ms
PAR  map: 6 ms
PAR time: 11 ms

希望能有所帮助。

Answer 2

@Fahim Bagar answer example应该在大数据集上运行得更快，但您应该先改进单线程代码，然后再草率地决定与并行相比的计时指标。

例如，通过@Fahim Bagar交换String[][] DataSet by Double[][] DataSet提供的代码示例可以很容易地删除浪费的Double.parseDouble

//String[][] DataSet = Matrix.map(mapping -> mapping.split(",")).toArray(String[][]::new);
Double[][] DataSet = Matrix.map(row -> Arrays.stream(row.split(",")).map(Double::parseDouble).toArray(Double[]::new)).toArray(Double[][]::new);

然后将DataSet[i + 1]和DataSet[j]的各种数组引用放在其循环外部的局部变量中：

for (int i = 0; i < distanceTable.length - 1; ++i) {
    Double[] arriplus1 = new Double[i + 1];
    Double[] iarr = DataSet[i + 1];
    for (int j = 0; j <= i; ++j) {
        double distance = 0.0;
        Double[] jarr = DataSet[j];
        for (int k = 0, sz = iarr.length; k < sz; ++k) {
            double difference = jarr[k] - iarr[k];
            distance += difference * difference;
        }
        arriplus1[j] = distance;
    }
    distanceTable[i] = arriplus1;
}

您可以对@Fahim Bagar euclidian方法执行相同的操作

public Double[] euclidian(Double[][] data) {
    Double[] result = new Double[data.length - 1];
    Double[] dL1 = data[data.length - 1];
    for (int i = 0; i < result.length; i++) {
        Double[] di = data[i];
        result[i] = Math.pow(di[0] - dL1[0], 2) + Math.pow(di[1] - dL1[1], 2);
    }
    return result;
}

在此之后，去掉Double并使用double将进一步加速/减少内存分配。

在CSV 1048行上，我在第10次运行每个行时看到了以下时间：

#####################
ORI read: 0 ms
ORI  map: 4 ms
ORI time: 14 ms
#####################
PAR read: 0 ms
PAR  map: 1 ms
PAR time: 10 ms