如何使用Spark和Java进行跨行集成？

Question

我目前正在尝试用Java编写一个Spark作业，用于计算数据集中列的积分。

数据如下所示：

    DateTime                velocity (in km/h)        vehicle 
    2016-03-28 11:00:45     80                        A
    2016-03-28 11:00:45     75                        A
    2016-03-28 11:00:46     70                        A
    2016-03-28 11:00:47     68                        A
    2016-03-28 11:00:48     72                        A
    2016-03-28 11:00:48     75                        A
    ... 
    2016-03-28 11:00:47     68                        B
    2016-03-28 11:00:48     72                        B
    2016-03-28 11:00:48     75                        B

为了计算每条线路的距离(以公里为单位)，我必须定义当前线路和下一条线路之间的时间差，并将其乘以速度。然后，必须将结果与前一行的结果相加，以检索当时行驶的“总距离”。

我现在想出了类似这样的东西。但它将计算每个地图作业的一辆车，并且可能有数百万条记录...

    final JavaRDD<String[]> input = sc.parallelize(Arrays.asList(
                new String[]{"2016-03-28", "11:00", "80", "VIN1"},
                new String[]{"2016-03-28", "11:00", "60", "VIN1"},
                new String[]{"2016-03-28", "11:00", "50", "VIN1"},
                new String[]{"2016-03-28", "11:01", "80", "VIN1"},
                new String[]{"2016-03-28", "11:05", "80", "VIN1"},
                new String[]{"2016-03-28", "11:09", "80", "VIN1"},
                new String[]{"2016-03-28", "11:00", "80", "VIN2"},
                new String[]{"2016-03-28", "11:01", "80", "VIN2"}
        ));

        // grouping by vehicle and date:
        final JavaPairRDD<String, Iterable<String[]>> byVinAndDate = input.groupBy(new Function<String[], String>() {
            @Override
            public String call(String[] record) throws Exception {
                return record[0] + record[3]; // date, vin
            }
        });

        // mapping each "value" (all record matching key) to result
        final JavaRDD<String[]> result = byVinAndDate.mapValues(new Function<Iterable<String[]>, String[]>() {
            @Override
            public String[] call(Iterable<String[]> records) throws Exception {
                final Iterator<String[]> iterator = records.iterator();

                String[] previousRecord = iterator.next();

                for (String[] record : records) {

                     // Calculate difference current <-> previous record
                     // Add result to new list

                    previousRecord = record;
                }

                return new String[]{
                        previousRecord[0],
                        previousRecord[1],
                        previousRecord[2],
                        previousRecord[3],
                        NewList.get(previousRecord[0]+previousRecord[1]+previousRecord[2]+previousRecord[2])

                };
            }
        }).values();

我完全不知道如何将这个问题转换为映射/减少转换，而不会失去分布式计算的好处。

我知道这违背了MR和Spark的本质，但是任何关于如何链接数据行或以优雅的方式解决这个问题的建议都会非常有帮助:)

谢谢!

Answer 1

我会说你做得对，你不应该害怕数百万张唱片：

• apache spark将做一个很好的工作来平衡它，一个工人可能忙于长任务，而另一个人可能处理几个较短的任务，
• 如果你可以解析时间和距离，那么你可能最终会得到双精度甚至整数，并且遍历数百万个双精度并不需要太多的担心。给定输入中的
• 不应该是数百万条记录，因为一天只有1440分钟。

虽然您的方法不需要任何额外的内存来计算，但我提出了另一种方法-使用aggregateByKey，并首先将所有时间和距离组合为每个键(vin，date)的数组。对于这个例子，我很抱歉，它是java 8。

    final JavaRDD<String[]> input = jsc.parallelize(Arrays.asList(
            new String[]{"2016-03-28", "11:00", "80", "VIN1"},
            new String[]{"2016-03-28", "11:00", "60", "VIN1"},
            new String[]{"2016-03-28", "11:00", "50", "VIN1"},
            new String[]{"2016-03-28", "11:01", "80", "VIN1"},
            new String[]{"2016-03-28", "11:05", "80", "VIN1"},
            new String[]{"2016-03-28", "11:09", "80", "VIN1"},
            new String[]{"2016-03-28", "11:00", "80", "VIN2"},
            new String[]{"2016-03-28", "11:01", "80", "VIN2"}
    ));

    input
            .mapToPair(v -> new Tuple2<>(v[0] + v[3], new Tuple2<>(v[1], v[2])))
            .aggregateByKey(
                    new Tuple2<>(new ArrayList<>(N), new ArrayList<>(N)),
                    (Tuple2<ArrayList<String>, ArrayList<String>> t, Tuple2<String, String> v) -> { //function to add new values to the collection
                        t._1().add(v._1());
                        t._2().add(v._2());
                        return t;
                    },
                    (Tuple2<ArrayList<String>, ArrayList<String>> t1, Tuple2<ArrayList<String>, ArrayList<String>> t2) -> { //function to combine collections
                        t1._1().addAll(t2._1());
                        t1._2().addAll(t2._2());
                        return t1;
                    })
            .foreach(v -> { //prints
                System.out.println();
                System.out.print(v);
            });

这段代码给我提供了以下内容

(2016-03-28VIN2,([11:00, 11:01],[80, 80]))
(2016-03-28VIN1,([11:00, 11:00, 11:00, 11:01, 11:05, 11:09],[80, 60, 50, 80, 80, 80]))

与在foreach中打印不同，您必须使用mapValues同时对两个数组进行循环，以获得与距离的差和乘法，然后使用reduceByKey((a, b) -> a + b)获得和。

为了节省一些内存并创建更少数量的ArrayLists，您可以在开头创建足够大的aggregateByKey，而不是N提供smth，如1000000，f.e.。

Answer 2

我更愿意将问题转换成dataframe API，使用spark，让spark来管理map/reduce (避免迭代器和数组)。实际上，我们想要计算每辆车/每段时间的距离。下面是我使用的步骤：

• 将RDD转换为数据帧

case类车辆(data: String，time: String，velocity: Int，id: String) val df = sc.parallelize(List(车辆(“2016-03-28”，"11:00"，80，"VIN1")，车辆(“2016-03-28”，"11:00"，60，"VIN1")，车辆(“2016-03-28”，"11:00"，50，"VIN1")，车辆(“2016-03-28”，"11:01"，80，"VIN1")，车辆(“2016-03-28”，"11:05"，80，"VIN1")，车辆(“2016-03-28”，"11:09"，80，"VIN1")，车辆(“2016-03-28”，"11:00"，80，"VIN2")，车辆(“2016-03-28”，“11:00”，80，“VIN2”)，"11:01"，80，"VIN2") .toDF()

• 由于某些数据被推送的时间相同(分钟:秒)，因此计算该时间的平均值(使用秒作为测量单位)

平均值速度=df.groupBy(df(“

”)，df("id")，df(“time”)).agg((avg(“速度”)/ 3600).as("avg_velocity"))

它将提供以下输出：

+----------+----+-----+--------------------+----+ | data| id| time| avg_velocity|rank| +----------+----+-----+--------------------+----+ |2016-03-28|VIN1|11:00|0.017592592592592594| 1| |2016-03-28|VIN1|11:01|0.022222222222222223| 2| |2016-03-28|VIN1|11:05|0.022222222222222223| 3| |2016-03-28|VIN1|11:09|0.022222222222222223| 4| |2016-03-28|VIN2|11:00|0.022222222222222223| 1| |2016-03-28|VIN2|11:01|0.022222222222222223| 2| +----------+----+-----+--------------------+----+

• 使用数据帧分析API计算时间范围，基于对数据和id列的分区，保留这些时间范围之间的秒数

val velocities = df.groupBy(df("data"), df("id"), df("time")).agg((avg("velocity") / 3600).as("avg\_velocity")) val overDataAndId = Window.partitionBy(df("data"), df("id")).orderBy(df("time")) val rank = denseRank.over(overDataAndId) val nextTime = lead(df("time"), 1).over(overDataAndId) val secondsBetween = udf((start: String, end: String) => { val sStart = time.LocalTime.parse(start) val sEnd = end match { case null => sStart case t: String if t.isEmpty => sStart case t: String if t.equalsIgnoreCase("null") => sStart case t: String => time.LocalTime.parse(end) } Seconds.secondsBetween(sStart, sEnd).getSeconds }) velocities.withColumn("rank", rank).show() velocities.withColumn("nextTime", nextTime).show() val seconds = velocities.withColumn("seconds", secondsBetween(df("time"), nextTime)) seconds.show()

它将输出：+----------+----+-----+--------------------+-------+ | data| id| time| avg_velocity|seconds| +----------+----+-----+--------------------+-------+ |2016-03-28|VIN1|11:00|0.017592592592592594| 60| |2016-03-28|VIN1|11:01|0.022222222222222223| 240| |2016-03-28|VIN1|11:05|0.022222222222222223| 240| |2016-03-28|VIN1|11:09|0.022222222222222223| 0| |2016-03-28|VIN2|11:00|0.022222222222222223| 60| |2016-03-28|VIN2|11:01|0.022222222222222223| 0| +----------+----+-----+--------------------+-------+

• 计算距离

的累加和

val distance = seconds.withColumn("distance", seconds("avg\_velocity") \* seconds("seconds")) distance.show() val cumulativeDistance = sum(distance("distance")).over(overDataAndId) val all = distance.withColumn("cum\_distance", cumulativeDistance) all.show()

它将输出累积距离(秒==为0的线路是每次每个车辆id的总距离)。在删除一些列之后，它将显示：

+----------+----+------------------+ | data| id| cum_distance| +----------+----+------------------+ |2016-03-28|VIN1|11.722222222222223| |2016-03-28|VIN2|1.3333333333333335| +----------+----+------------------+

我发现它是一种更具可读性的解决方案，它让spark管理数据帧上的操作。代码是用scala编写的，但是可以很容易地翻译成java。