如何将行与 Spark 和 Java 集成？答案

【问题标题】：How to integrate over rows with Spark and Java?如何将行与 Spark 和 Java 集成？
【发布时间】：2016-04-06 14:39:37
【问题描述】：

我目前正在尝试用 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 的本质背道而驰，但任何有关如何互连数据行或以优雅方式解决此问题的建议都会非常有帮助:)

谢谢！

【问题讨论】：

标签： java hadoop apache-spark rdd integral

【解决方案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]))

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

为了节省一些内存并创建更少数量的 ArrayList，您可以在开始时创建足够大的它们 - aggregateByKey 的第一行 - 而不是 N 提供 1000000 之类的东西，f.e.

【讨论】：

【解决方案2】：

我宁愿将问题转换为数据帧 API，使用 spark，让 spark 管理 map/reduce（避免迭代器和数组）。实际上，我们想要计算每个车辆/每个时间段的距离。所以这是我使用的步骤：

将 RDD 转换为数据帧

case class Vechicle(data: String, time: String, velocity: Int, id: String)
    val df = sc.parallelize(List(
        Vechicle("2016-03-28", "11:00", 80, "VIN1"),
        Vechicle("2016-03-28", "11:00", 60, "VIN1"),
        Vechicle("2016-03-28", "11:00", 50, "VIN1"),
        Vechicle("2016-03-28", "11:01", 80, "VIN1"),
        Vechicle("2016-03-28", "11:05", 80, "VIN1"),
        Vechicle("2016-03-28", "11:09", 80, "VIN1"),
        Vechicle("2016-03-28", "11:00", 80, "VIN2"),
        Vechicle("2016-03-28", "11:01", 80, "VIN2")
      )).toDF()

因为某些数据是在同一时间（分钟：秒）推送的，所以计算平均值（使用秒作为度量单位）

val 速度 = df.groupBy(df("data"), df("id"), df("time")).agg((avg("velocity") / 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 计算时间帧，基于对 data 和 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 翻译。

【讨论】：