Spark - 如何正确处理RDD.map()方法中的错误情况？

Question

我正在尝试使用Spark RDD进行一些文本处理.

输入文件的格式为:

2015-05-20T18:30 <some_url>/?<key1>=<value1>&<key2>=<value2>&...&<keyn>=<valuen>

我想从文本中提取一些字段并将它们转换为CSV格式,如:

<value1>,<value5>,<valuek>,<valuen>

以下代码是我如何做到这一点:

val lines = sc.textFile(s"s3n://${MY_BUCKET}/${MY_FOLDER}/test/*.gz")
val records = lines.map { line =>
    val mp = line.split("&")
                 .map(_.split("="))
                 .filter(_.length >= 2)
                 .map(t => (t(0), t(1))).toMap

    (mp.get("key1"), mp.get("key5"), mp.get("keyk"), mp.get("keyn"))
}

我想知道,如果输入文本的某些行格式错误或无效,则该map()函数无法返回有效值.这在文本处理中应该很常见,处理这个问题的最佳做法是什么？

Answer 1

为了管理这些错误,您可以使用scala的类在flatMap操作中尝试,在代码中:

    val lines = sc.textFile(s"s3n://${MY_BUCKET}/${MY_FOLDER}/test/*.gz")
    val records = lines.flatMap (line =>
        Try{
          val mp = line.split("&")
              .map(_.split("="))
              .filter(_.length >= 2)
              .map(t => (t(0), t(1))).toMap

          (mp.get("key1"), mp.get("key5"), mp.get("keyk"), mp.get("keyn"))
      } match {
        case Success(map) => Seq(map)
        case _ => Seq()
    })

有了这个,你只有"好的",但如果你想要两个(错误和好的),我建议使用一个返回Scala Either的map函数,然后在代码中使用Spark过滤器:

    val lines = sc.textFile(s"s3n://${MY_BUCKET}/${MY_FOLDER}/test/*.gz")
    val goodBadRecords = lines.map (line =>
        Try{
          val mp = line.split("&")
              .map(_.split("="))
              .filter(_.length >= 2)
              .map(t => (t(0), t(1))).toMap

          (mp.get("key1"), mp.get("key5"), mp.get("keyk"), mp.get("keyn"))
      } match {
        case Success(map) => Right(map)
        case Failure(e) => Left(e)
    })
    val records = goodBadRecords.filter(_.isRight)
    val errors = goodBadRecords.filter(_.isLeft)

我希望这会有用