如何在Spark数据集中显示（或操作）由Kryo编码的对象？

2024-05-21 • 问答

说你有这个：

// assume we handle custom type
class myobj(val i: Int,val j: String)
implicit val myobjEncoder = org.apache.spark.sql.Encoders.kryo[myobj]
val ds = spark.createDataset(Seq(new myobj(1,"a"),new myobj(2,"b"),new myobj(3,"c")))

做ds.show时，我得到：

+--------------------+
|               value|
+--------------------+
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
+--------------------+

我知道这是因为内容被编码为内部Spark SQL二进制表示形式。但是如何显示这样的解码内容？

+---+---+
| _1| _2|
+---+---+
|  1|  a|
|  2|  b|
|  3|  c|
+---+---+

UPDATE1

显示内容不是最大的问题，更重要的是在处理数据集时可能会导致问题，请考虑以下示例：

// continue with the above code
val ds2 = spark.createDataset(Seq(new myobj(2,new myobj(6,new myobj(5,"c"))) 

ds.joinWith(ds2,ds("i") === ds2("i"),"inner") 
// this gives a Runtime error: org.apache.spark.sql.AnalysisException: Cannot resolve column name "i" among (value);

这是否意味着kryo编码类型不能方便地执行类似joinWith的操作？然后我们如何处理Dataset上的自定义类型？

scala> println(spark.version) 3.0.1 // Note that case keyword case class MyObj(val i: Int,val j: String) import org.apache.spark.sql.Encoders implicit val myObjEncoder = Encoders.kryo[MyObj] // myObjEncoder: org.apache.spark.sql.Encoder[MyObj] = class[value[0]: binary] val ds = (Seq(new MyObj(1,"a"),new MyObj(2,"b"),new MyObj(3,"c"))).toDS // the Kryo deserializer gives bytes scala> ds.printSchema root |-- value: binary (nullable = true) scala> :type sc org.apache.spark.SparkContext // Let's deserialize the bytes into an object import org.apache.spark.serializer.KryoSerializer val ks = new KryoSerializer(sc.getConf) // that begs for a generic UDF val deserMyObj = udf { value: Array[Byte] => import java.nio.ByteBuffer ks.newInstance.deserialize(ByteBuffer.wrap(value)).asInstanceOf[MyObj] } val solution = ds.select(deserMyObj('value) as "result").select($"result.*") scala> solution.show +---+---+ | i| j| +---+---+ | 1| a| | 2| b| | 3| c| +---+---+

如何在Spark数据集中显示（或操作）由Kryo编码的对象？

zhaojun730730 回答：如何在Spark数据集中显示（或操作）由Kryo编码的对象？

大家都在问