如何在Spark DataFrame中从Mapstruct提取数据？

Question

如何从df_raw中提取数据（标签）这是Mapstruct？

我正在使用Spark 1.6。我在Spark中通过hivesql从Hive获取数据，然后我得到了一个数据框，但数据框中的一列是Mapstruct，我试图从中提取数据，但失败了，非常希望stackoverflow能给我一些帮助，3Q。

从Hive获取数据后，我得到了一个名为df\u raw的数据帧，模式为：

root
 |-- subscriberid: string (nullable = true)
 |-- time: string (nullable = true)
 |-- itemid: string (nullable = true)
 |-- label: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- partitiondate: string (nullable = true)

和df\U raw。显示（3）为：

+------------+-------------------+------+--------------------+-------------+
|subscriberid|               time|itemid|               label|partitiondate|
+------------+-------------------+------+--------------------+-------------+
|     1569960|2019-09-08 08:00:01| 46611|Map(license -> yo...|     20190908|
|     1104555|2019-09-08 08:00:29| 46445|Map(license -> wa...|     20190908|
|     1309036|2019-09-08 08:00:55| 45219|Map(license -> yo...|     20190908|
+------------+-------------------+------+--------------------+-------------+

为了清楚地获得它，我将df\u raw转换为rdd，并从中获取2个数据：

val rawRDD: RDD[String] = df_raw.rdd.map(pojo => pojo.mkString("\t"))
println("——————————" + "\n")
rawRDD.take(2).foreach(println)

数据为：

1545807 2019-09-10 07:29:41 4706    Map(license -> wa, videoid -> 4706, mediapaytype -> 1, duration -> 131) 20190908
1496840 2019-09-10 07:30:43 4535    Map(license -> you, videoid -> 4535, mediapaytype -> 1, duration -> 137)    20190908

我想知道如何分别从df\U原始列（“标签”）中提取数据？

我试图得到这样一个新的数据帧：

  val df_userBehaviorsRow_1 = rawUserBehaviorsData.map(line => {
    val splits = line.split("\t")

    val subscriberid = splits(0)
    val time= splits(1)
    val itemid = splits(2)

    val label = splits(3)
    val resultant = label.map{m=>
      val seq=m.values.toSeq
      (seq(0),seq(1),seq(2))
    }

    val license = resultant._1
    val duration = resultant._3

    (subscriberid , time, itemid, label, license，duration)
  }).toDF

我失败了，IntelliJ IDEA甚至不能识别“val结果=label.map{m=

非常希望能得到一些帮助，3Q。

Answer 1

例如，为了选择列中的许可证值，您只需选择该列并应用密钥许可证。

import org.apache.spark.sql.functions.sql.col
df_raw.select(col("label")("license")).show()

您可以使用withColumn将列许可证添加到数据帧

 df_raw_new = df_raw.withColumn("license", col("label")("license").alias("license"))