python - Spark MapPartitionRDD 无法打印值

Question

我正在关注《Machine Learning with Spark Book》，并尝试将 python 代码转换为 scala 代码，并使用 Beaker 笔记本共享变量，以便将值传递给 python 以使用 matplotlib 进行绘图，如书中所述。到目前为止，我已经能够转换大部分代码，但在使用 u.item 数据集进行数据清理的 try-catch 转换方面遇到了一些问题。下面的代码以无限循环结束，没有明确的问题是什么错误。

val movieData = sc.textFile("/Users/minHenry/workspace/ml-100k/u.item")
val movieDataSplit = movieData.first()

val numMovies = movieData.count()

def convertYear(x:String):Int = x.takeRight(4) match {
         case x => x.takeRight(4).toInt
         case _ => 1900
    }

val movieFields = movieData.map(lines => lines.split('|'))
print(movieData.first())
val years1 = movieFields.map(fields => fields(2))

val years = movieFields.map(fields => fields(2).map(x=>convertYear(x.toString())))
val filteredYears = years.filter(x => x!=1900)
years.take(2).foreach(println)

我怀疑我的问题出在我的模式匹配上，但我不确定它出了什么问题。我认为 takeRight() 有效，因为它不会提示该函数所应用的类型。

更新

根据迄今为止提供的答案的建议，我已更新代码如下:

import scala.util.Try
val movieData = sc.textFile("/Users/minHenry/workspace/ml-100k/u.item")

def convertYear(x:String):Int = Try(x.toInt).getOrElse(1900)
val movieFields = movieData.map(lines => lines.split('|'))

val preYears = movieFields.map(fields => fields(2))
val years = preYears.map(x => x.takeRight(4))//.map(x=>convertYear(x))
println("=======> years")
years.take(2).foreach(println) //--output = 1995/n1995
println("=======> filteredYears")
val filteredYears = years.filter(x => x!=1900)

filteredYears.take(2).foreach(println)
//val movieAges = filteredYears.map(yr => (1998-yr)).countByValue() 

我注释掉了 takeRight(4) 后面的 map，因为它比放置 x=>convertYear(x.takeRight(4)) 更容易注释 并且应该产生相同的输出。当我应用这个 convertYear() 函数时，我仍然陷入无限循环。这些值按照所显示的几个打印语句中的预期打印。问题是，如果我无法删除无法轻松转换为 Int 的数据点，那么我无法运行最后一行中的 countByValue() 函数。

以下是我的公共(public)烧杯笔记本的链接，以了解更多背景信息: https://pub.beakernotebook.com/#/publications/56eed31d-85ad-4728-a45d-14b3b08d673f

Answer 1

1. movieData: RDD[String]
2. movieFields: RDD[Array[String]]
3. years1: RDD[String]
4. val years = movieFields.map(fields => fields(2).map(x=>convertYear(x.toString()))) -fields(2)是 String等等x是 Char ，因为String被视为 Seq[Char] 。所有输入convertYear(x: String)只有一个字母字符串。

您的错误是类型不兼容隐藏( convertYear(x.toString()) )。这是警钟。始终在 scala 中使用类型系统，不要隐藏 toString() 的问题或isInstanceOf或者是其他东西。然后编译器在运行之前显示错误。

附注

1. 第二次调用takeRight是没用的。
2. def convertYear(x:String):Int = x.takeRight(4) match { case x => x.takeRight(4).toInt case _ => 1900 }

模式匹配是关于检查类型或条件(使用 if 语句)。您的第一个部分功能不会检查任何内容。所有输入都转到x.takeRight(4).toInt 。也无法防御 toInt异常(exception)。 使用 def convertYear(x: String): Int = Try(x.toInt).getOrElse(1900) 代替.

更新

scala> import scala.util.Try
import scala.util.Try

scala> def convertYear(x:String):Int = Try(x.toInt).getOrElse(1900)
convertYear: (x: String)Int

scala> List("sdsdf", "1989", "2009", "1945", "asdf", "455")
res0: List[String] = List(sdsdf, 1989, 2009, 1945, asdf, 455)

scala> res0.map(convertYear)
res1: List[Int] = List(1900, 1989, 2009, 1945, 1900, 455)

与 RDD 都一样，因为它是一个像 List 一样的仿函数。

val filteredYears = years.filter(x => x!=1900)不会像你期望的那样工作。 x是一个 String 而不是 Int。 Scala 不会隐式转换类型以进行比较。所以你总是得到 true .