为具有复杂列的DataFrame添加新列（数组＜映射＜字符串，字符串＞）

Question

我正在使用以下模式从外部源加载数据帧:

 |-- 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，其中包含列类型数组和以下case类：

case类META_DATA_COL（M_1： String，M_2： String，M_3，M_10： String）

基于示例，我这里的方法是创建一个UDF并传递元数据列。但是由于它是一个复杂的类型，我在解析它时遇到了很多困难。

除此之外，对于UDF中的“new”变量M_10，我还需要对方法进行一些字符串操作。所以我需要访问元数据列中的每个元素。

处理这个问题的最佳方式是什么？我试图将源数据帧（元数据）转换为case类；但这不起作用，因为它在进入UDF时被转换回spark 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*/)

这应该会有帮助，如果你能提供更多的处理细节，我可以提供更多的帮助。