join - 加入spark时如何跳过空rdd

Question

我想从 Cassandra 获得 2 rdd，然后加入他们。我想跳过空值。

def extractPair(rdd: RDD[CassandraRow]) = {
    rdd.map((row: CassandraRow) => {

     val name = row.getName("name")
     if (name == "")
         None   //join wrong
     else
        (name, row.getUUID("object"))

    })
  }

  val rdd1 = extractPair(cassRdd1)
  val rdd2 = extractPair(cassRdd2)
  val joinRdd = rdd1.join(rdd2)  //"None" join wrong

使用 flatMap 可以解决这个问题，但我想知道如何使用地图解决这个问题

def extractPair(rdd: RDD[CassandraRow]) = {
        rdd.flatMap((row: CassandraRow) => {

         val name = row.getName("name")
         if (name == "")
             seq()
         else
            Seq((name, row.getUUID("object")))

        })
      }

Answer 1

这是不可能的map。您需要使用filter. 但是您仍然最好将有效结果包装在Some. 但是，那么你仍然会将它包裹在 Some 结果中......需要一秒钟map才能打开它。所以，实际上，你最好的选择是这样的：

def extractPair(rdd: RDD[CassandraRow]) = {
  rdd.flatMap((row: CassandraRow) => {
    val name = row.getName("name")
    if (name == "") None
    else Some((name, row.getUUID("object")))
  })
}

Option可以隐式转换为可展平的类型，并更好地传达您的方法消息。