Добавление нового столбца для DataFrame со сложным столбцом (Array>

Question

Я загружаю Dataframe из внешнего источника со следующей схемой:

 |-- A: string (nullable = true)
 |-- B: timestamp (nullable = true)
 |-- C: long (nullable = true)
 |-- METADATA: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- M_1: integer (nullable = true)
 |    |    |-- M_2: string (nullable = true)
 |    |    |-- M_3: string (nullable = true)
 |    |    |-- M_4: string (nullable = true)
 |    |    |-- M_5: double (nullable = true)
 |    |    |-- M_6: string (nullable = true)
 |    |    |-- M_7: double (nullable = true)
 |    |    |-- M_8: boolean (nullable = true)
 |    |    |-- M_9: boolean (nullable = true)
 |-- E: string (nullable = true)

Теперь мне нужно добавить новый столбец METADATA_PARSED с типом столбца Array и следующиеcase case:

case case META_DATA_COL (M_1: String, M_2: String, M_3, M_10: String )

Мой подход, основанный на примерах, заключается в созданииUDF и передать в столбце METADATA. Но так как он сложного типа, у меня много проблем с его синтаксическим анализом.

Кроме того, в UDF для "новой" переменной M_10 мне нужно выполнить некоторые строковые манипуляции с методомтакже. Поэтому мне нужно получить доступ к каждому из элементов в столбце метаданных.

Как лучше всего подойти к этой проблеме? Я попытался преобразовать исходный фрейм данных (+ METADATA) в класс case;но это не сработало, поскольку при входе в UDF оно было переведено обратно в типы WrappedArray.

Answer 1

Вы можете использовать что-то вроде этого.

import org.apache.spark.sql.functions._

val tempdf  = df.select(
  explode( col("METADATA")).as("flat") 
)

val processedDf = tempdf.select( col("flat.M_1"),col("flat.M_2"),col("flat.M_3"))

Теперь напишите udf

def processudf = udf((col1:Int,col2:String,col3:String) => /* do the processing*/)

, это должно помочь, я могу предоставить дополнительную помощь, если вы можете предоставить более подробную информацию ообработка.