Scala RDD 操作答案

【问题标题】：Scala RDD OperationScala RDD 操作
【发布时间】：2017-08-15 05:09:55
【问题描述】：

我是 scala 的新手。

我有一个 csv 文件存储在 hdfs 中。我正在使用 scala 读取该文件

 val salesdata = sc.textFile("hdfs://localhost:9000/home/jayshree/sales.csv")

这里是“销售”数据的一个小样本。

C_ID    T_ID    ITEM_ID ITEM_Price
5       199        1       500
33      235        1       500
20      249        3       749
35      36         4       757
19      201        4       757
17      94         5       763
39      146        5       763
42      162        5       763
49      41         6       824
3       70         6       824
24      161        6       824
48      216        6       824

我必须对其执行以下操作。

1.对每件商品应用一些折扣，在 d(itemprice) 列上假设 30% 的折扣。公式为 d=d-(30%(d))。

2.在对每件商品应用 30% 的折扣后，找出客户明智的最小和最大商品价值。

我尝试将 30 与 ITEM_Price 列的观察值相乘。问题是 d 的值被视为字符串。当我在结果中乘以一个数字时，它会多次显示该值。比如 (500*3 = 500500500)

我可以将其转换为数据框并执行此操作。但我只是想知道，如果不将其转换为数据帧，我们是否可以对 RDD 进行这些操作。

【问题讨论】：

我试图将该列提取为数组[int]。但是，如果我对数组的每个元素进行任何算术运算，则会引发一些错误，例如 * 不是数组 [int] 的一部分。我不需要完整的代码。我只需要一个指南来获得结果。我需要遵循哪些步骤。

标签： scala apache-spark rdd

【解决方案1】：

折扣

case class Transaction(cId: Int, tId: Int, itemId: Int, itemPrice: Int)

val salesdata : RDD[String]=> 映射 RDD，在映射中用分隔符拆分行，然后将 Array 转换为名为 Transaction 的案例类，调用 Array(i).toInt 来转换字段。在这一步中，您的目标是获得一个 RDD[Transaction]。
再次映射 RDD 并应用折扣复制您的交易 (t => t.copy(itemPrice=0.7*t.itemPrice))
您将拥有一个新的 RDD[Transaction]

以客户为中心

获取最后一个对象，应用 keyBy(_.cId) 以获取 RDD[Int, Transaction]，其中您的密钥是客户端。
通过键减少添加每个项目的价格。目标 => RDD[Int, Int] 获取每个客户的总数。
找到您的目标客户！

【讨论】：

我已使用此代码将该 rdd 转换为数组 " val discount = salesdata.map( str => str.split(",")).map( c => (c(0) .toInt, c(1).toInt, c(2).toInt, c(3).toDouble))”。当我在数组上使用案例类时，出现错误。
salesdata.map(str => str.split(",")).map(c => Transaction(c(0).toInt, c(1).toInt, c(2) .toInt, c(3).toDouble)) 你的错误是什么？
引起：java.lang.NumberFormatException：对于输入字符串：C_ID"
当我运行以下代码时，你能帮我吗？我没有收到任何错误，但是当我尝试将 rdd 保存为文本文件时，我遇到了错误。我不明白代码哪里错了。 1. val salesdata = sc.textFile("hdfs://localhost:9000/home/jayshree/sales.csv") 2. val salesdata_arr = salesdata.map(str => str.split(",")).map ( c => (c(0).toInt, c(1).toInt, c(2).toInt, c(3).toDouble)) 3.salesdata_arr.saveAsTextFile("hdfs://localhost:9000/home /jayshree/discount_output")
问题是你的csv有标题行，所以你不能将字符串“C_ID”转换为整数。您可以将 spark-csv 库用于 spark-sql 或删除标头 stackoverflow.com/questions/27854919/…

【解决方案2】：

既然您需要更多指南，让我们先看看 Spark 之外的内容，然后将其视为典型的 Scala 集合。

您的数据将如下所示：

val data = Array(
        (5, 199, 5, 100),
        (33, 235, 5, 100),
...
)

我认为使用split 或正则表达式之类的，将salesdata 字符串RDD 映射到Array 或Tuple4 的RDD 不会有任何问题。

让我们使用一个元组。然后你可以这样做：

data.map {
  case (cId, tId, item, price) => (cId, tId, item, price * .7)
}

这将元组的原始 RDD 映射到另一个元组 RDD，其中最后一个值，即价格，减少了 30%。所以结果是Tuple4[Int, Int, Int, Double]。

说实话，我不知道您所说的客户方面的最小值和最大值是什么意思，但也许是这样的：

data.map {
  case (cId, tId, item, price) => (cId, tId, item, price * .7)
}.groupBy(_._1)
.mapValues { tuples => 
  val discountedPrices = tuples.map(_._4)
  (discountedPrices.min, discountedPrices.max)
}

首先，我做一个groupBy，它从cId（元组中的第一个值，它解释._1）到一个完整元组的集合产生一个Map——所以一个Map的cId 到与cId 相关的行集合。在 Spark 中，这将产生一个PairRDD。

Map 和 PairRDD 都有一个 mapValues 函数，它允许我在转换每个元组集合时保留键（cIds）。在这种情况下，我只需通过获取每个元组中的第 4 个项目（折扣价格）将集合映射到折扣价格集合。然后我在该集合上调用 min 和 max 并返回这些值的元组。

所以结果是客户 ID 的 Map 到折扣价格的最小值和最大值的元组。 RDD API 的美妙之处在于它如此紧密地遵循传统的 Scala 集合 API，所以它基本上是一回事。

【讨论】：