如何使用 Stream Parallel Java 执行矩阵计算？答案

【问题标题】：How Do I Perform Matrix Calculation With Stream Parallel Java?如何使用 Stream Parallel Java 执行矩阵计算？
【发布时间】：2020-02-23 17:04:12
【问题描述】：

我正在尝试使用多维数组（[verybigrow][2]）制作矩阵算术运算方法。我是新手，我只是找不到我做错了什么。非常感谢任何帮助告诉我它是什么。

    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 );
        }

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

数据处理的输出应该是这样的：

[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]

【问题讨论】：

发生的事情不是你所期望的？
是的，如果在小数据中运行，该程序就可以工作。但是如果最大数据超过 100 万，则会出现内存堆错误
我在您的代码中看不到 multithreading、parallel-processing、bigdata。请不要添加不必要的标签。
首先：了解java命名约定。变量和字段名称采用驼峰命名法。然后：不要从并行流开始。如果有的话，只需使用普通流。以最直接最简单的方式解决您的问题。只有当这行得通，并且您了解您正在做的所有事情时，然后才开始研究接下来的步骤。然后：请告诉我们您的矩阵计算应该得到什么结果。不要将您的代码扔给我们，以便我们确定您在那里尝试做什么。告诉我们。
*叹息* 是的，当然。首先将您的String DataSet[][] 转换为Double DataSet[][]，然后使用值进行计算，而不是调用parseDouble 百万次。您会看到计算速度快了约 10 倍，没有并行化。如果您想要一个高效的并行解决方案，那么当前的答案当然也不是最好的方法。

标签： java multithreading parallel-processing

【解决方案1】：

@Fahim Bagar 答案示例在使用大数据集时应该运行得更快，但您应该先改进单线程代码，然后再仓促决定与并行相比的时序指标。

例如，使用@Fahim Bagar 提供的代码示例将String[][] DataSet 替换为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

【讨论】：

【解决方案2】：

我尝试将matrixDistance 更改为distanceTable。尝试将此代码移动到不同的方法中，以便您可以并行运行它

        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

希望对你有帮助。

【讨论】：

感谢您对问题的评论。你可以分享代码sn-p run吗？或存储库代码，以便我可以使用我的实际数据自行测试。
查看我的repository。不客气。
非常感谢
谢谢fahim，我已经测试了你的代码。它适用于Ruspini Dataset。速比很远。非常感谢。希望这对我的研究有所帮助。
我担心测试结果会产生误导，因为您的测试程序只在 7 行上运行一次计时，而 JIT 将对更大的数据集产生巨大的影响。首先运行 PAR 或 ORI 计时中的哪个时间最长 - 尝试交换调用 main.parallel(file) 和 main.original(file) 的部分。在我的旧 PC (2 CORE) 上，当您比较最后执行的版本的时间时，ORI 代码总是比并行代码快。看看并行版本在多大的测试数据下变得最快会很有趣。