Spark Scala scala.util.control.Exception 在地图中捕获和丢弃 None

Question

我正在 scala 中编写一个 Spark 应用程序，并希望处理一个脏输入文件。

// CSV file
val raw_data = sc.textFile(...)

val clean_data = raw_data.map(_.split(delimiter))
  .map( r => (r(0), r(1).toDouble)

当 r(1) 不是数字时会抛出 NumberFormatException。这发生在丑陋的输入数据中的少数行上。

我终于找到了一种丑陋的方式来完成我需要的事情：

import scala.util.control.Exception._

val clean_data = raw_data.map(_.split(delimiter))
  .map( r => (r(0),
        catching(classOf[NumberFormatException]).opt(r(1).toDouble))
  .filter( r => r._2 != None)
  .map( r => (r._1, r._2.get))

这给我留下了两个问题。

1) 在地图中简单地删除格式错误的行的最佳方法是什么？

2) 如何处理通过 catch 创建的 Option 类型，而无需先显式过滤掉 None，然后将 .get 函数映射并应用于非 None Option 值？

我尝试应用 .flatMap(identity) 步骤来摆脱 Nones，但得到了预期的：TraversableOnce[?] 异常。

Answer 1

在 Spark 中 collect(pf:PartialFunction) 是 scala 集合 collect 的孪生兄弟，其存在正是为了这个目的：保留那些在偏函数中定义的集合元素。

val rawData = sc.textFile(...)

val cleanData = rawData.map(_.split(Delimiter))
             .collect{ case Array(x,y) if (Try(y.toDouble).isSuccess) (x,y.toDouble) }

另一个不评估 .toDouble 两次的选项是使用 flatMap：

val cleanData = rawData.map(_.split(Delimiter))
                       .flatMap(entry => Try(entry.toDouble).toOption)

注意：在 Spark 中有点令人困惑的是，有一个无参数的 collect 方法用于将数据从 RDD 获取到驱动程序。