我无法让Spark将json
(或csv)作为一个带有Option[_]
字段的case类的Dataset
读取,其中并非所有字段在源中都被定义。
有点晦涩,但是假设我有一个名为CustomData
的case类。
给定以下json文件(customA.json
):
{"id":123, "colA": "x", "colB": "z"}
{"id":456, "colA": "y"}
{"id":789, "colB": "a"}
以下是代码:
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.master("local[2]")
.appName("test")
.getOrCreate()
import spark.implicits._
case class CustomData(id: BigInt, colA: Option[String], colB: Option[String])
org.apache.spark.sql.catalyst.encoders.OuterScopes.addOuterScope(this)
val ds = spark
.read
.option("mode", "PERMISSIVE")
.json("src/main/resources/customA.json")
.as[CustomData]
.show()
输出结果如预期一样:
+----+----+---+
|colA|colB| id|
+----+----+---+
| x| z|123|
| y|null|456|
|null| a|789|
+----+----+---+
虽然不是所有的列都必须被定义。 但是如果我想使用相同的代码来读取一个没有某个列出现的文件,我无法实现这一点:
对于另一个json文件(customB.json
):
{"id":321, "colA": "x"}
{"id":654, "colA": "y"}
{"id":987}
还有额外的代码:
val ds2 = spark
.read
.option("mode", "PERMISSIVE")
.json("src/main/resources/customB.json")
.as[CustomData]
.show()
输出结果为错误信息:
org.apache.spark.sql.AnalysisException: 给定输入列[colA, id]无法解析'colB';
这很有道理,但我希望能够重复使用同一种情况类来处理两个文件。特别是如果在摄取之前我不知道
colB
是否出现在JSON文件中。当然,我可以进行检查,但是否有一种方法将不存在的列转换为
null
(与customA.json
一样)?将readmode设置为Permissive
似乎没有改变任何内容。我错过了什么吗?
map/select
代替foldLeft/withColumn
,切换到了Encoders.product[T].schema
而不是ScalaReflection.schemaFor[T]
,并将dataFrame.columns
移到了一个惰性变量上以避免重新评估。 - Tom Lous