展平 RDD 中的嵌套元组

Question

我正在使用 Spark SQL 从表中提取行。其中一些数据是重复出现的，我正在尝试计算出现的次数。本质上，我正在尝试执行基本的“字数统计”示例，但我的数据不是以下形式：(Word : String, Count : Int)，而是有一行数据替换了单词/字符串。

更具体地说，我的数据看起来像：RDD[((row), count)]，其中行是从 sql 表中提取的，包含字符串、双精度数、整数等。

它是RDD 形式，因为我想使用reduceByKey。请参阅：Avoid groupByKey。它是一个 (Key, Value) 对，有一个很长的键（来自 sql 数据库的某行），其值为“字数”。

我的应用正在这样做：

myDataframe
    // Append a 1 to each row
    .map(row => (row, 1))
    // Convert to RDD so we can use the reduceByKey method
    .rdd
    // Add up the 1's corresponding to matching keys
    .reduceByKey(_ + _)
    //Filter by rows that show up more than 10 times
    .filter(_._2 > 100)

    ...

现在假设我的行数据包含(string, double, int)。 这是我想将我的数据从RDD[((string, double, int), count)] 解包到RDD[(string, double, int, count)] 的地方，以便我最终可以将这些数据保存到另一个 SQL 表中。

有没有什么方法可以让我解压这个...嵌套元组...之类的内容？

我的解决方案是像这样“解包” RDD 的元素： .map(row => (row._1._1, row._1._2, row._1._3, row._2))

但一定有更好的方法！如果我决定从行中获取更多元素，则必须修改此 .map() 调用。

谢谢！

Answer 1

您可以使用Row 的toSeq 和fromSeq，如下例所示：

val df = Seq(
  ("a", 10.0, 1),
  ("a", 10.0, 1),
  ("b", 20.0, 2),
  ("c", 30.0, 3),
  ("c", 30.0, 3)
).toDF("c1", "c2", "c3")

import org.apache.spark.sql.Row

df.rdd.
  map((_, 1)).
  reduceByKey(_ + _).
  filter(_._2 > 1).
  map{
    case (row: Row, count: Int) => Row.fromSeq(row.toSeq :+ count)
  }
// res1: Array[org.apache.spark.sql.Row] = Array([a,10.0,1,2], [c,30.0,3,2])

Answer 2

您不必重新使用 RDD；您正确引用的文章警告不要使用RDD.groupByKey，但它不应该应用于DataFrame 的groupBy。在 DataFrame 上使用 groupBy 是安全的（并且性能良好）！查看更多here。

因此，要按所有 DataFrame 的列进行分组，计算每个组的出现次数，并过滤计数 > 10 的组，您可以简单地使用：

df.groupBy(df.columns.map(col): _*) // alternatively: df.groupBy(df.columns.head, df.columns.tail: _*)
  .count()
  .filter($"count" > 10)

结果具有类似于输入的架构，并带有额外的count 长列。