如何有效地对 spark 数据集中的每 k 行进行分组？答案

【问题标题】：How to efficiently group every k rows in spark dataset?如何有效地对 spark 数据集中的每 k 行进行分组？
【发布时间】：2018-11-05 00:28:15
【问题描述】：

我创建了一个 spark Dataset[Row]，Row 是 Row(x: Vector)。 x 这里是一个 1xp 向量。

是否可以 1) 将每 k 行分组 2) 将这些行连接成一个 k x p 矩阵 - mX 即，将 Dateset[Row(Vector)] 更改为 Dateset[Row(Matrix)] ？

这是我目前的解决方案，将此 Dataset[Row] 转换为 RDD，并使用 zipWithIndex 和 aggregateByKey 连接每 k 行。

val dataRDD = data_df.rdd.zipWithIndex
    .map {  case (line, index) =>  (index/k, line) }
    .aggregateByKey(...) (..., ...)

但是好像效率不是很高，有没有更高效的方法呢？

提前致谢。

【问题讨论】：

标签： apache-spark dataset rdd

【解决方案1】：

您的方法存在两个性能问题：

使用全局排序
洗牌以建立k 的组

如果您绝对需要全局排序，从第 1 行开始，并且您不能将数据分成多个分区，那么 Spark 必须通过单个核心移动所有数据。您可以通过找到一种拥有多个分区的方法来加快该部分的速度。

您可以通过使用mapPartitions 一次处理一个分区的数据来避免洗牌：

spark.range(1, 20).coalesce(1).mapPartitions(_.grouped(5)).show

+--------------------+
|               value|
+--------------------+
|     [1, 2, 3, 4, 5]|
|    [6, 7, 8, 9, 10]|
|[11, 12, 13, 14, 15]|
|    [16, 17, 18, 19]|
+--------------------+

请注意，上面的 coalesce(1) 将所有 20 行强制为一个分区。

【讨论】：

嗨，Sim，谢谢您的评论。我实际上不需要全球订购。 k 组也不必要地需要改组。我将 rdd.repartition(...).mapPartitions(_.grouped(5)) 应用于我的代码以避免全局分组。我只是想知道上面的代码是否会发生两次改组？即，一个用于重新分区，另一个用于 mapPartiion。
repartition() 会随机播放，mapPartitions() 不会。不过，除非您需要分发数据，否则您不需要 repartition()。 coalesce() 不会随机播放。

【解决方案2】：

这是一个将 N 条记录分组为列的解决方案：

从RDD生成到DF，如下图处理。

g 是组，k 是记录 g 内重复的编号的关键。 v 是您的记录内容。

输入是一个 6 行的文件，我在这里使用了 3 组。

唯一的缺点是行的余数小于分组 N。

import org.apache.spark.sql.functions._
import org.apache.spark.mllib.rdd.RDDFunctions._

val dfsFilename = "/FileStore/tables/7dxa9btd1477497663691/Text_File_01-880f5.txt"
val readFileRDD = spark.sparkContext.textFile(dfsFilename)
val rdd2 = readFileRDD.sliding(3,3).zipWithIndex
val rdd3 = rdd2.map(r => (r._1.zipWithIndex, r._2))
val df = rdd3.toDF("vk","g")

val df2 = df.withColumn("vke", explode($"vk")).drop("vk")
val df3 = df2.withColumn("k", $"vke._2").withColumn("v", $"vke._1").drop("vke")

val result = df3
            .groupBy("g")
            .pivot("k")
            .agg(expr("first(v)"))

result.show()

+---+--------------------+--------------------+--------------------+
|  g|                   0|                   1|                   2|
+---+--------------------+--------------------+--------------------+
|  0|The quick brown f...|Here he lays I te...|Gone are the days...|
|  1|  Gosh, what to say.|Hallo, hallo, how...|          I am fine.|
+---+--------------------+--------------------+--------------------+

【讨论】：