json的Spark异常处理

Question

我正在尝试在读取json文件时捕获/忽略解析错误

val DF = sqlContext.jsonFile("file")

有几行不是有效的json对象,但是数据太大而无法单独进行(~1TB)

我遇到了映射使用import scala.util.Try和in.map(a => Try(a.toInt))引用的异常处理: 如何在spark map()函数中处理异常？

使用该函数读取json文件时如何捕获异常sqlContext.jsonFile？

谢谢!

Answer 1

不幸的是你在这里不走运。DataFrameReader.json在引擎盖下使用的几乎是全有或全无。如果您的输入包含格式错误的行，您必须手动过滤它们。一个基本的解决方案可能如下所示：

import scala.util.parsing.json._

val df = sqlContext.read.json(
    sc.textFile("file").filter(JSON.parseFull(_).isDefined)
)

由于上述验证相当昂贵，您可能更愿意完全删除jsonFile/read.json并直接使用解析的 JSON 行。