将 RDD 拆分为多个 RDD 和 Cache答案

【问题标题】：Split RDD into many RDDs and Cache将 RDD 拆分为多个 RDD 和 Cache
【发布时间】：2017-05-09 19:42:29
【问题描述】：

我也有这样的 rdd

(aid, session, sessionnew, date)
(55-BHA, 58, 15, 2017-05-09)
(07-YET, 18, 5, 2017-05-09)
(32-KXD, 27, 20, 2017-05-09)
(19-OJD, 10, 1, 2017-05-09)
(55-BHA, 1, 0, 2017-05-09)
(55-BHA, 19, 3, 2017-05-09)
(32-KXD, 787, 345, 2017-05-09)
(07-YET, 4578, 1947, 2017-05-09)
(07-YET, 23, 5, 2017-05-09)
(32-KXD, 85, 11, 2017-05-09)

我想将具有相同辅助的所有内容拆分为一个新的 rdd，然后将其缓存以供以后使用，因此每个唯一辅助一个 rdd。我看到了其他一些答案，但他们正在将 rdds 保存到文件中。在内存中保存这么多 rdds 有问题吗？可能会在 30k+ 左右

我用 spark jobserver 保存缓存的 rdd。

【问题讨论】：

标签： scala apache-spark apache-spark-sql spark-jobserver

【解决方案1】：

我建议你cachegrouped rdd如下
假设你有 rdd 数据：

val rddData = sparkContext.parallelize(Seq(
      ("55-BHA", 58, 15, "2017-05-09"),
      ("07-YET", 18, 5, "2017-05-09"),
      ("32-KXD", 27, 20, "2017-05-09"),
      ("19-OJD", 10, 1, "2017-05-09"),
      ("55-BHA", 1, 0, "2017-05-09"),
      ("55-BHA", 19, 3, "2017-05-09"),
      ("32-KXD", 787, 345, "2017-05-09"),
      ("07-YET", 4578, 1947, "2017-05-09"),
      ("07-YET", 23, 5, "2017-05-09"),
      ("32-KXD", 85, 11, "2017-05-09")))

您可以cache将数据与“aid”分组并使用filter选择您需要的grouped data：

val grouped = rddData.groupBy(_._1).cache
val filtered = grouped.filter(_._1 equals("32-KXD"))

但我建议您使用DataFrame，如下所示，它比rdds 更有效和改进

import sqlContext.implicits._
val dataFrame = Seq(
  ("55-BHA", 58, 15, "2017-05-09"),
("07-YET", 18, 5, "2017-05-09"),
("32-KXD", 27, 20, "2017-05-09"),
("19-OJD", 10, 1, "2017-05-09"),
("55-BHA", 1, 0, "2017-05-09"),
("55-BHA", 19, 3, "2017-05-09"),
("32-KXD", 787, 345, "2017-05-09"),
("07-YET", 4578, 1947, "2017-05-09"),
("07-YET", 23, 5, "2017-05-09"),
("32-KXD", 85, 11, "2017-05-09")).toDF("aid", "session", "sessionnew", "date").cache

val newDF = dataFrame.select("*").where(dataFrame("aid") === "32-KXD")
newDF.show

希望对你有帮助

【讨论】：