如何创建键值 RDD (Scala)

Question

我有以下类型的 RDD（名称：AllTrainingDATA_RDD） org.apache.spark.rdd.RDD[(String, Double, Double, String)] :

(ICCH_1,4.3,3.0,Iris-setosa)
(ICCH_1,4.4,2.9,Iris-setosa)
(ICCH_1,4.4,3.0,Iris-setosa)
(ICCH_2,4.4,3.2,Iris-setosa)

第一列：ICCH_ID，第二列：X_Coordinates，第三列：Y_Coordinates，第四列：类

我想最终得到一个 RDD，它有 第 2 列和第 3 列作为键，第 4 列作为值。 ICCH_ID 列应保留在 RDD 中。

我目前基于一些互联网研究的尝试是这样的：

val AllTrainingDATA_RDD_Final = AllTrainingDATA_RDD.map(_.split(",")).keyBy(_(X_COORD,Y_COORD)).mapValues(fields => ("CLASS")).groupByKey().collect()

但是我得到了这个错误：

错误：值拆分不是 (String, Double, Double, String) 的成员

附：我正在使用 Databricks 社区版。我是 Scala 新手。

Answer 1

让我们尝试逐步分解您的解决方案：

val AllTrainingDATA_RDD_Final = AllTrainingDATA_RDD
    .map(_.split(","))
    .keyBy(_(X_COORD,Y_COORD))
    .mapValues(fields => ("CLASS"))
    .groupByKey()
    .collect()

你的第一个问题是.map(_.split(","))的使用。这可能是在 RDD[String] 上完成的预处理阶段，以从文本输入行中提取逗号分隔值。但既然你已经这样做了，我们可以继续删除该部分。

你的第二个问题来自.keyBy(_(X_COORD,Y_COORD))，它看起来像这样：

错误：(String, Double, Double, String) 不带参数

这是因为您提供了 keyBy 一个匿名函数，该函数试图在 RDD 中的每个元组上应用 (X_COORD,Y_COORD)，但您真正想要的是提取 x 和 y 坐标（第二和第三值）的函数从你的元组。实现此目的的一种方法是.keyBy{case (_, x, y, _) => (x, y)}

最后，您使用mapValues 只会为RDD 中的所有元素生成相同的字符串值("CLASS")。相反，您可以像这样简单地取元组中的第 4 个值：.mapValues(_._4)

将所有这些放在一起，您会得到以下代码：

val AllTrainingDATA_RDD_Final = AllTrainingDATA_RDD
    .keyBy{case (_, x, y, _) => (x, y)}
    .mapValues(_._4)
    .groupByKey()
    .collect()

由于您是 Scala 新手，我建议您在继续之前花一些时间熟悉语法、特性和 API。它将帮助您更快地理解和克服这些问题。