Apache spark scala 异常处理答案

【问题标题】：Apache spark scala Exception handlingApache spark scala 异常处理
【发布时间】：2017-12-20 08:30:24
【问题描述】：

如何在 Spark 中进行异常处理 - Scala 中的无效记录这是我的代码：

val rawData = sc.textFile(file)
val rowRDD = rawData.map(line => Row.fromSeq(line.split(",")))
val rowRDMapped = rowRDD.map { x => x.get(1), x.get(10) }
val DF = rowRDMapped.toDF("ID", "name" )

如果输入数据正常，一切正常，如果我没有足够的字段，我会得到 ArrayIndexOutOfBoundException。

我正在尝试使用 try-catch，但我无法通过 try catch 跳过包含无效数据的记录

val rowRDMapped = rowRDD.map { try {
                                    x => x.get(1), x.get(10) 
                                    }catch {
                                        println("Invalid Data")
                                        //Here it expects to return ROW, but I am not sure what to do here, since I dont want any data to be returned.
                                    }
                             }

请让我知道如何使用 try catch 解决问题，如果有更好的解决方案，那也会有很大帮助

【问题讨论】：

你想要什么？跳线太短？
@Zernike，感谢您的关注，是的，我需要跳过短行，这会导致 Index out of bound 异常。

标签： scala apache-spark exception-handling try-catch

【解决方案1】：

最简单的：

val rawData = sc.textFile(file)
val rowRDD = rawData.map(line => Row.fromSeq(line.split(",")))
val rowRDMapped = rowRDD.filter(_.length >= 11).map(x => x.get(1), x.get(10))

最好使用collect（不要与other function混淆）

val rowRDMapped = rowRDD.collect{x if x.length >= 11 => x.get(1), x.get(10)}

【讨论】：

答案无关紧要。真正的要求可能是任何的。对于此示例，正确的是 11，因为 Row 具有从零开始的索引。
是的，你是对的。我也打算写11，但我错误地写了9。:)谢谢

【解决方案2】：

您可以使用下面的 try catch 并稍后过滤

val rawData = sc.textFile(file)
val rowRDD = rawData.map(line => Row.fromSeq(line.split(",")))
val rowRDMapped = rowRDD.map(x => (Try(x.get(1).toString) getOrElse "blank", Try(x.get(10).toString) getOrElse "blank"))
val DF = rowRDMapped.toDF("ID", "name").filter($"name" =!= "blank")

【讨论】：

【解决方案3】：

你可以使用Try代替try-catch

下面的代码将过滤掉没有足够字段的数据行，并获取带有剩余字段的数据帧。

val rawData = sc.textFile(line)
val rowRDD = rawData.map(line => Row.fromSeq(line.split(",")))
val rowRDMapped = rowRDD.flatMap{ x => Try(x.getString(1), x.getString(10)).toOption }
val DF = rowRDMapped.toDF("ID", "name")

【讨论】：