我想将以 Json 形式读入的 Dataframe 转换为给定类的 Dataset。到目前为止,当我能够编写自己的案例类时,效果很好。
case class MyCaseClass(...)
val df = spark.read.json("path/to/json")
val ds = df.as[MyCaseClass]
def myFunction(input: MyCaseClass): MyCaseClass = {
// Do some validation and things
input
}
ds.map(myFunction)
但是,现在我绑定(bind)到外部 Java 类(特别是由 thrift 创建的类)。所以这里有一个更具体的自定义类示例:
JSON:
{"a":1,"b":"1","wrapper":{"inside":"1.1", "map": {"k": "v"}}}
{"a":2,"b":"2","wrapper":{"inside":"2.1", "map": {"k": "v"}}}
{"a":3,"b":"3","wrapper":{"inside":"3.1", "map": {"k": "v"}}}
类:
class MyInnerClass(var inside: String, var map: Map[String, String]) extends java.io.Serializable {
def getInside(): String = {inside}
def setInside(newInside: String) {inside = newInside}
def getMap(): Map[String, String] = {map}
def setMap(newMap: Map[String, String]) {map = newMap}
}
class MyClass(var a: Int, var b: String, var wrapper: MyInnerClass) extends java.io.Serializable {
def getA(): Int = {a}
def setA(newA: Int) {a = newA}
def getB(): String = {b}
def setB(newB: String) {b = newB}
def getWrapper(): MyInnerClass = {wrapper}
def setWrapper(newWrapper: MyInnerClass) {wrapper = newWrapper}
}
所以我想做的是:
val json = spark.read.json("path/to/json")
json.as[MyClass]
但是,抛出:
Unable to find encoder for type stored in a Dataset. Primitive type (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases.
import org.apache.spark.sql.Encoders
val kryoMyClassEncoder = Encoders.kryo[MyClass]
json.as[MyClass](kryoMyClassEncoder)
抛出:
Try to map struct<a:bigint,b:string,wrapper:struct<inside:string,map:struct<k:string>>> to Tuple1, but failed as the number of fields does not line up
那么如何将 Dataframe 转换为自定义对象 Dataset。
最佳答案
不要使用 kryo 编码器,而是尝试使用产品编码器,即:
val productMyClassEncoder = Encoders.product[MyClass]
关于java - Spark Dataframe 到 Java 类的 Dataset,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41876965/