Scala + Spark - 如何从“scala.collection.mutable.WrappedArray$ofRef”转换为我们的自定义对象？

Question

我们在项目中使用spark 和Scala。

我们正在使用自定义encoders 来创建spark datasets。

我们的数据集架构类型如下：

(String, util.ArrayList[MyObject])

当我们执行 df.printSchema 时，我们得到以下信息：

root
 |-- key: string (nullable = true)
 |-- listOfMyObject: array (nullable = true)
 |    |-- element: binary (containsNull = true)

MyObject 是如下的 scala 案例类：

case class MyObject(key: String, dataList: java.util.ArrayList[MyObject2])

当我们在这个数据框上应用地图函数时，

df.map((row) => {
val key  = row.get(0)
val values = row.get(1)
})

在运行时，行包含以下架构：

StructField(key,StringType,true)
StructField(myObject,ArrayType(BinaryType,true),true)

我们能够检索到String 值，但是在尝试检索util.ArrayList[MyObject] 时，我们得到了scala.collection.mutable.WrappedArray$ofRef。

我们使用ref.getClass 方法得到了这个。

有什么方法可以解决这个问题吗？

谢谢

阿努杰

Answer 1

尝试使用spark可以通过编码器管理的Scala对象，在您的情况下，您可以将java.util.ArrayList更改为scala的简单List，然后如果列的顺序和类型正确，则转换仅使用以下简单行将您的数据框转换为您的类类型的数据集：

case class MyObject(key: String, dataList: List[MyObject2])
val ds: Dataset[MyObject] = df.as[MyObject]

请记住，如果架构有数组，则只能将其转换为列表，不能转换为数组。

Answer 2

我会将java.util.ArrayList 更改为scala.collection.immutable.List

package playground

import org.apache.spark.sql.SparkSession
// import java.util.ArrayList

object Casting {

  val spark = SparkSession.builder()
    .appName("Casting")
    .config("spark.master", "local[*]")
    .getOrCreate()

  case class MyObject2(fname: String, age: Int)

  // case class MyObject(key: String, dataList: ArrayList[MyObject2])
  case class MyObject(key: String, dataList: List[MyObject2])

  // val arrLst = new util.ArrayList[MyObject2]()
  // arrLst.add(MyObject2("Marie", 25))
  // arrLst.add(MyObject2("Peter", 27))

  val lst = List(MyObject2("Marie", 25),MyObject2("Peter", 27))

  // val data: Seq[MyObject] = Seq(MyObject("key",arrLst))
  val data: Seq[MyObject] = Seq(MyObject("key",lst))

  def main(args: Array[String]): Unit = {

    import spark.implicits._

    try {
      val df = spark.createDataFrame(data)

      df.printSchema()
      df.map(row => {
        val key  = row.key
        val values = row.dataList
        (key, values)
      }).show(false)

      df.show()
    } finally {
      spark.stop()
      println("Spark Session has stopped.")
    }
  }
}

root
 |-- key: string (nullable = true)
 |-- dataList: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- fname: string (nullable = true)
 |    |    |-- age: integer (nullable = false)

+-----+--------------------------+
|_1   |_2                        |
+-----+--------------------------+
|Lucia|[[Marie, 25], [Peter, 27]]|
+-----+--------------------------+

ArrayList 不是Scala collection 的成员。 我会尝试与Scala objects 合作，Spark 可以通过其encoders 管理。