如何在 Scala 的地图函数中使用 if 进行过滤？

Question

我有一个从 sc.newAPIHadoopFile 生成的 hadoopFiles 对象。

scala> hadoopFiles
res1: org.apache.spark.rdd.RDD[(org.apache.hadoop.io.LongWritable, org.apache.hadoop.io.Text)] = UnionRDD[64] at union at <console>:24

我打算遍历 hadoopFiles 中的所有行并对其进行操作和过滤，其中应用了 if 检查并将抛出异常：

scala> val rowRDD = hadoopFiles.map(line =>
     |           line._2.toString.split("\^") map {
     |             field => {
     |               var pair = field.split("=", 2)
     |               if(pair.length == 2)
     |                 (pair(0) -> pair(1))
     |             }
     |           } toMap
     |         ).map(kvs => Row(kvs("uuid"), kvs("ip"), kvs("plt").trim))
<console>:33: error: Cannot prove that Any <:< (T, U).
                 } toMap
                   ^

但是，如果我删除 if(pair.length == 2) 部分，它将正常工作：

scala>     val rowRDD = hadoopFiles.map(line =>
     |           line._2.toString.split("\^") map {
     |             field => {
     |               var pair = field.split("=", 2)
     |               (pair(0) -> pair(1))
     |             }
     |           } toMap
     |         ).map(kvs => Row(kvs("uuid"), kvs("ip"), kvs("plt").trim))
warning: there was one feature warning; re-run with -feature for details
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.catalyst.expressions.Row] = MappedRDD[66] at map at <console>:33

谁能告诉我这种现象的原因，并告诉我正确的应用if语句的方法。非常感谢！

P.S. 我们可以使用这个简化的例子来测试：

"1=a^2=b^3".split("\^") map {
            field => {
              var pair = field.split("=", 2)
              if(pair.length == 2)
                pair(0) -> pair(1)
              else
                return
            }
          } toMap

Answer 1

要映射集合并仅保留部分映射元素，您可以使用 flatMap。 flatMap 采用 returns 集合的函数，例如实例 Option。现在 if 表达式需要有一个 else 部分 returns 一个空的 Option，即 None.

scala> val rowRDD = hadoopFiles.map(line =>
     |           line._2.toString.split("\^") flatMap {
     |             field => {
     |               var pair = field.split("=", 2)
     |               if (pair.length == 2)
     |                 Some(pair(0) -> pair(1))
     |               else
     |                 None
     |             }
     |           } toMap
     |         ).map(kvs => Row(kvs("uuid"), kvs("ip"), kvs("plt").trim))

Answer 2

您可以使用 collect:

val res = "1=a^2=b^3".split("\^") collect {
  _.split("=", 2) match {
    case Array(a, b) => a -> b
  }
} toMap

println(res) // Map(1 -> a, 2 -> b)

在您的特定情况下，会发生以下情况：

case class Row(uuid: String, ip: String, plt: String)
val hadoopFiles = List(("", "uuid=a^ip=b^plt"))

val rowRDD = hadoopFiles.map(line =>
  line._2.toString.split("\^") map {
    field =>
      {
        var pair = field.split("=", 2)
        val res = if (pair.length == 2)
          (pair(0) -> pair(1))
        res  // res: Any (common super class for (String, String)
             // which is Tuple2 and Unit (result for case when 
             // pair.length != 2)
      }
  } /* <<< returns Array[Any] */ /*toMap*/ ) 
  //.map(kvs => Row(kvs("uuid"), kvs("ip"), kvs("plt").trim))

inner map 的结果是 Any 并且 map 产生 Array[Any]。如果您查看 toMap 定义，您将看到：

  def toMap[T, U](implicit ev: A <:< (T, U)): immutable.Map[T, U] = {
    val b = immutable.Map.newBuilder[T, U]
    for (x <- self)
      b += x // <<< implicit conversion from each `x` of class `A` in `self`
             // to (T, U) because we have `implicit ev: A <:< (T, U)`
    b.result()
  }

对于您的 Array[Any]，在当前上下文中没有从 Any 到 (T, U) 的隐式转换。因此，您的代码失败了。如果您添加 else 备选方案：

  val rowRDD = hadoopFiles.map(line =>
    (line._2.toString.split("\^") map {
      field =>
        {
          var pair = field.split("=", 2)
          val res = if (pair.length == 2)
            (pair(0) -> pair(1))
          else ("" -> "") // dummy, just for demo
          res // res: (String, String)
        }
    } toMap) withDefaultValue ("") 
           /*withDefaultValue just to avoid Exception for this demo*/ )
    .map(kvs => Row(kvs("uuid"), kvs("ip"), kvs("plt").trim))

  println(rowRDD) // List(Row(a,b,))

此处您的结果将是 Array[(String, String)] 并且存在从 (String, String) 到 (T, U) 的隐式转换。因此代码可以编译并运行。

如何在 Scala 的地图函数中使用 if 进行过滤？

How to use if to filter in a map function in scala?

scala

collections

apache-spark

flatmap