从UDF中的SparkSQL行中提取嵌套数组

Question

我正在处理数据帧，需要提取数据。我有许多嵌套的级别，所以我使用分解和选择来创建第一个级别，但随后我对嵌套级别使用UDF。

我有一个UDF，它取Root.Obj，这是一个数组，我希望它返回一个数组[MyObj]
我的输出类：

case class MyObj(fieldA: Boolean, fieldB: String, fieldC: Array[MyNested])
case class MyNested(field1: Long, field2: String)

简而言之，这是输入模式：

 |-- Obj: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- FieldA: boolean (nullable = true)
 |    |    |-- FieldB: string (nullable = true)
 |    |    |-- FieldC: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- Field1: long (nullable = true)
 |    |    |    |    |-- Field2: string (nullable = true)
 |    |    |-- FieldD: boolean (nullable = true)

我的自定义项：

def extractObjs: UserDefinedFunction = udf {
  (objs: Seq[Row]) ⇒
    objs.map {
      obj ⇒
        MyObj(
          obj.getAs[Boolean]("FieldA"),
          obj.getAs[String]("FieldB"),
          extractNested(obj.get???("FieldC"))
        )
    }
}

def extractNested(nesteds: ???): Array[MyNested] = {
  ???
}

这是更复杂的IRL，因为我需要从其他地方检索值，并且有更多的嵌套数组。此外，Obj和FieldC的输入结构比这里复杂得多，我不能（或不想）为它们创建case类。因为我需要在多个地方这样做，所以假设我不知道FieldC元素的“结构”。

我的问题是提取“FieldC”数组。我想要一个Seq[Row]，但我无法实现，getSTRt只给我一个Row，getSeq[Row]在抛出错误之后，因为scala.collection.mutable.WrapedArray$ofRef不能强制转换为org.apache.spark.sql.Row。

Answer 1

结构映射到UDF中的行，因此可以通过以下方式访问结构数组：

def extractObjs: UserDefinedFunction = udf {
  (objs: Seq[Row]) ⇒
    objs.map {
      obj ⇒
        MyObj(
          obj.getAs[Boolean]("FieldA"),
          obj.getAs[String]("FieldB"),
          extractNested(obj.getAs[Seq[Row]]("FieldC"))
        )
    }
}

def extractNested(nesteds: Seq[Row]): Array[MyNested] = {
  nesteds.map(r => MyNested(r.getAs[Long]("Field1"),r.getAs[String]("Field2"))).toArray
}