映射和删除Spark Scala中的重复项?

Sai*_*Sai 3 scala scala-collections apache-spark apache-spark-sql

我有一个数据集test.txt。它包含如下数据

1::1::3
1::1::2
1::2::2
2::1::5
2::1::4
2::2::2
3::1::1
3::2::2
Run Code Online (Sandbox Code Playgroud)

我已经使用以下代码创建了数据框。

case class Rating(userId: Int, movieId: Int, rating: Float)
def parseRating(str: String): Rating = {
val fields = str.split("::")
assert(fields.size == 3)
Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat)
}

val ratings = spark.read.textFile("C:/Users/cravi/Desktop/test.txt").map(parseRating).toDF()
Run Code Online (Sandbox Code Playgroud)

但是当我尝试打印输出时,我的输出低于输出

[1,1,3.0]
[1,1,2.0]
[1,2,2.0]
[2,1,2.0]
[2,1,4.0]
[2,2,2.0]
[3,1,1.0]
[3,2,2.0]
Run Code Online (Sandbox Code Playgroud)

但是我想打印输出如下,即删除重复的组合,而不是field(2) value 1.0

[1,1,1.0]
[1,2,1.0]
[2,1,1.0]
[2,2,1.0]
[3,1,1.0]
[3,2,1.0]
Run Code Online (Sandbox Code Playgroud)

Ram*_*jan 5

创建dataframe删除副本后,可以通过调用来删除重复项,.dropDuplicates(columnNames)1.0可以通过使用litwithColumn函数来填充第三列。

综上所述,简单的解决方案可以如下

val ratings = sc.textFile("C:/Users/cravi/Desktop/test.txt")
    .map(line => line.split("::"))
    .filter(fields => fields.length == 3)
    .map(fields => Rating(fields(0).toInt, fields(1).toInt, 1.0f))
  .toDF().dropDuplicates("userId", "movieId")

ratings.show(false)
Run Code Online (Sandbox Code Playgroud)

这应该给你

+------+-------+------+
|userId|movieId|rating|
+------+-------+------+
|3     |1      |1.0   |
|2     |2      |1.0   |
|1     |2      |1.0   |
|1     |1      |1.0   |
|2     |1      |1.0   |
|3     |2      |1.0   |
+------+-------+------+
Run Code Online (Sandbox Code Playgroud)