在Spark数据集中编码ujson.Value

Question

假设我将这些JSON行存储在一个文本文件中。

{"a": "...", "data": [{}]}
{"a": "...", "data": [{"b": "..."}]}
{"a": "...", "data": [{"d": "..."}]}
{"a": "...", "data": [{"b": "...", "c": "..."}]}

我希望将该文件处理为Spark Dataset，但我不知道字段data的确切模式。我使用upickle将JSON转换为case类。

case class MyCC(a: String, data: Seq[ujson.Value.Obj])

implicit val r: Reader[MyCC] = macroR

sc.textFile("s3://path/to/file.txt")
  .map(uread[MyCC](_))
  .toDS                 // Dataset[MyCC]
  .show()

尝试这样做，我得到了以下错误：

java.lang.UnsupportedOperationException: No Encoder found for ujson.Value
- map value class: "ujson.Value"
- field (class: "scala.collection.mutable.LinkedHashMap", name: 
"value")
- array element class: "ujson.Obj"
- field (class: "scala.collection.Seq", name: "data")
- root class: "com.mycaule.MyCC"

我如何解决这个数据建模问题？

谢谢

Answer 1

我可以根据需要读取数据，而不需要创建自定义编码器。我只是正确地定义了case类。

import scala.collection.mutable
case class CustomClass( a: String,
                        data: List[mutable.HashMap[String,String]]
                              )

val dataSourceName =  "s3/path/to/data.json"

val schema = ScalaReflection.schemaFor[CustomClass].dataType.asInstanceOf[StructType]

val data = spark.read.schema(schema).json(dataSourceName).as[CustomClass]

data.show(10, truncate = false)

以下是输出：

+---+----------------------+
|a  |data                  |
+---+----------------------+
|...|[[]]                  |
|...|[[b -> ...]]          |
|...|[[d -> ...]]          |
|...|[[b -> ..., c -> ...]]|
+---+----------------------+