慢火花应用程序-java

Question

我正在尝试创建一个 spark 应用程序，它采用lat、long、时间戳点的数据集，并在网格单元格内增加单元格计数。网格由lon、lat和time作为 z 轴的 3d 单元组成。

现在我已经完成了应用程序并且它做了它应该做的，但是扫描整个数据集需要几个小时（~9g）。我的集群由 3 个节点组成，每个节点有 4 个核心，每个 8g 内存，我目前使用 6 个执行器，每个执行器有 1 个核心和 2g。

我猜我可以对代码进行相当多的优化，但是我的代码中是否存在导致这种延迟的大错误？

    //Create a JavaPairRDD with tuple elements. For each String line of lines we split the string 

//and assign latitude, longitude and timestamp of each line to sdx,sdy and sdt. Then we check if the data point of 

//that line is contained in a cell of the centroids list. If it is then a new tuple is returned

//with key the latitude, Longitude and timestamp (split by ",") of that cell and value 1.

    JavaPairRDD<String, Integer> pairs = lines.mapToPair(x -> {

        String sdx = x.split(" ")[2];

        String sdy = x.split(" ")[3];

        String sdt = x.split(" ")[0];

        double dx = Double.parseDouble(sdx);

        double dy = Double.parseDouble(sdy);

        int dt = Integer.parseInt(sdt);

        List<Integer> t = brTime.getValue();

        List<Point2D.Double> p = brCoo.getValue();

        double dist = brDist.getValue();

        int dur = brDuration.getValue();

        for(int timeCounter=0; timeCounter<t.size(); timeCounter++) {

            for ( int cooCounter=0; cooCounter < p.size(); cooCounter++) {

                double cx = p.get(cooCounter).getX();

                double cy = p.get(cooCounter).getY();

                int ct = t.get(timeCounter);

                String scx = Double.toString(cx);

                String scy = Double.toString(cy);

                String sct = Integer.toString(ct);

                if (dx > (cx-dist) && dx <= (cx+dist)) {

                    if (dy > (cy-dist) && dy <= (cy+dist)) {

                        if (dt > (ct-dur) && dt <= (ct+dur)) {

                            return new Tuple2<String, Integer>(scx+","+scy+","+sct,1);

                        }

                    }

                }

            }

        }

        return new Tuple2<String, Integer>("Out Of Bounds",1);

    });

Answer 1

尝试使用 mapPartitions 速度更快，请参阅此示例链接；另一件事是将这部分代码放在循环 timeCounter 之外

Answer 2

可能导致运行这样的 Spark 地图的成本的最大因素之一与 RDD 上下文之外的数据访问有关，这意味着驱动程序交互。根据你的情况，也有变量至少4个存取这种情况出现：brTime，brCoo，brDist，和brDuration。看来您正在通过String#split而不是利用内置程序进行一些行解析。最后，scx,&nbsp;scy, 和sct都是为每个循环计算的，尽管只有在它们的数字对应项通过一系列检查时才会返回它们，这意味着浪费了 CPU 周期和额外的 GC。如果不实际审查工作计划，很难说上述是否会使绩效达到可接受的水平。查看您的历史服务器应用程序日志，看看是否有任何阶段占用了您的时间——一旦您确定了其中的罪魁祸首，这就是实际需要优化的地方。

Answer 3

我尝试了 mappartitionstopair 并且还移动了 scx、scy 和 sct 的计算，以便仅当点通过条件时才计算它们。应用速度大幅提升仅17分钟！我相信 mappartitionsopair 是最大的因素。