spark mapPartitionRDD 无法打印值答案

【问题标题】：spark mapPartitionRDD can't print valuesspark mapPartitionRDD 无法打印值
【发布时间】：2016-03-21 23:34:47
【问题描述】：

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

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() 函数。

这里是我的公共烧杯笔记本的链接以获取更多上下文： https://pub.beakernotebook.com/#/publications/56eed31d-85ad-4728-a45d-14b3b08d673f

【问题讨论】：

你的convertYear方法应该做什么？
显然有一个错误的数据点，所以它用硬编码值 1900 替换它，然后这个值被完全删除。此函数转换的字段格式为“Jan-15-1988”或类似的日期值

标签： python scala apache-spark

【解决方案1】：

movieData: RDD[String]
movieFields: RDD[Array[String]]
years1: RDD[String]
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 或其他东西的问题。然后编译器在运行前显示错误。

附：

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

模式匹配是关于检查类型或条件（使用 if 语句）。您的第一个部分功能不检查任何内容。所有输入都转到x.takeRight(4).toInt。也没有针对toInt 异常的防御。请改用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 是字符串而不是 Int。 Scala 不会隐式转换类型以进行比较。所以你总是得到true。

【讨论】：