慢火花应用程序 - java

Question

我正在尝试创建一个 spark 应用程序，它采用 lat、long、timestamp 点的数据集，如果它们增加单元格计数位于网格单元内。网格由以 lon、lat 和 time 为 z 轴的 3d 单元格组成。

现在我已经完成了应用程序，它完成了它应该做的事情，但是扫描整个数据集需要几个小时（~9g）。我的集群由 3 个节点组成，每个节点有 4 个核心，每个 8g ram，我目前使用 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、scy 和 sct 都是针对每个循环计算的，尽管它们只有在它们的数字对应物通过一系列检查时才会返回，这意味着浪费了 CPU 周期和额外的 GC。

如果不实际审查工作计划，很难说以上是否会使绩效达到可接受的水平。查看您的历史服务器应用程序日志，看看是否有任何阶段占用了您的时间 - 一旦您确定了其中的罪魁祸首，这就是真正需要优化的地方。

Answer 3

我尝试了 mappartitionstopair 并且还移动了 scx、scy 和 sct 的计算，以便仅当点满足条件时才计算它们。应用程序的速度仅用了17分钟就得到了显着提升！我相信 mappartitionsopair 是最大的因素。非常感谢 Mks 和 bsplosion！