scala - Spark Scala scala.util.control.Exception 在 map 中捕获并删除 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) 简单删除 map 中格式错误的行的最佳方法是什么？

2) 我如何处理通过捕获创建的选项类型，而不需要首先显式过滤掉 None 然后映射并在非 None 选项值上应用 .get 函数？

我尝试应用 .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 获取数据到驱动程序。