scala – Apache Spark：处理RDD中的Option / Some / None

发布时间：2020-12-16 18:45:38 所属栏目：安全来源：网络整理

导读：我正在映射HBase表,每个HBase行生成一个RDD元素.但是,有时行有坏数据(在解析代码中抛出NullPointerException),在这种情况下我只想跳过它. 我有我的初始映射器返回一个Option,表示它返回0或1个元素,然后筛选Some,然后获取包含的值： // myRDD is RDD[(Immutab

我正在映射HBase表,每个HBase行生成一个RDD元素.但是,有时行有坏数据(在解析代码中抛出NullPointerException),在这种情况下我只想跳过它.

我有我的初始映射器返回一个Option,表示它返回0或1个元素,然后筛选Some,然后获取包含的值：

// myRDD is RDD[(ImmutableBytesWritable,Result)]
val output = myRDD.
  map( tuple => getData(tuple._2) ).
  filter( {case Some(y) => true; case None => false} ).
  map( _.get ).
  // ... more RDD operations with the good data

def getData(r: Result) = {
  val key = r.getRow
  var id = "(unk)"
  var x = -1L

  try {
    id = Bytes.toString(key,11)
    x = Long.MaxValue - Bytes.toLong(key,11)
    // ... more code that might throw exceptions

    Some( ( id,( List(x),// more stuff ...
        ) ) )
  } catch {
    case e: NullPointerException => {
      logWarning("Skipping id=" + id + ",x=" + x + "; n" + e)
      None
    }
  }
}

有没有更惯用的方法来缩短这个？我觉得这看起来很乱,在getData()和我正在做的map.filter.map舞蹈中.

也许flatMap可以工作(产生在序列0或1项),但我不希望它弄平我创建的地图功能的元组,只是消除清空.

解决方法

如果您更改getData以返回scala.util.Try,那么您可以大大简化转换.像这样的东西可以工作：

def getData(r: Result) = {
  val key = r.getRow
  var id = "(unk)"
  var x = -1L

  val tr = util.Try{
    id = Bytes.toString(key,11)
    // ... more code that might throw exceptions

    ( id,( List(x)
          // more stuff ...
     ) )
  } 

  tr.failed.foreach(e => logWarning("Skipping id=" + id + ",x=" + x + "; n" + e))
  tr
}

然后你的变换可以这样开始：

myRDD.
  flatMap(tuple => getData(tuple._2).toOption)

如果您的Try是一个失败,它将通过toOption变为None,然后作为flatMap逻辑的一部分被删除.此时,转换中的下一步将仅处理成功的情况,即无需从包含的getData返回的基础类型(即无选项)

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!