查找数组列的最大值并在数据框中查找另一个数组中的关联值

Question

我有一个包含以下数据的 csv 文件。

Id	Subject	Marks
1	M,P,C	10,8,6
2	M,P,C	5,7,9
3	M,P,C	6,7,4

我需要在每个 Id 的标记列中找出最大值，并从主题列中找到关联主题。

我想要的结果应该是：

Id	Subject	Marks
1	M	10
2	C	9
3	P	7

我正在读取 csv 文件并使用逗号值拆分将主题和标记作为数组列。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Column

 val spark = SparkSession.builder().getOrCreate()
    import spark.implicits._
    
    val df = spark.read.format("CSV")
                  .option("header", "true")
                  .option("delimiter", "|")
                  .option("inferSchema", "true")
                  .load("file:///p:/test/Data/test.csv")

   val df1 = df.select(col("id"),
                        split(col("subjects"),",").as("subjects"),
                        split(col("Makrs"),",").as("Makrs")
                      )

   df1.printSchema()

df1 模式是：

root
 |-- id: integer (nullable = true)
 |-- Sub: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- Mark: array (nullable = true)
 |    |-- element: string (containsNull = true)

而 df1 数据是；

+---+---------+----------+
| id| subjects|     Makrs|
+---+---------+----------+
|  1|[M, P, C]|[10, 8, 6]|
|  2|[M, P, C]| [5, 7, 9]|
|  3|[M, P, C]| [6, 7, 4]|
+---+---------+----------+

我不知道如何在数据框中的数组列中找到最大值。

我试过 array_max 但得到一个错误未找到：值 array_max

df1.withColumn("MaxMarks", array_max($"Makrs")).show()

Answer 1

如果您将 Dataframe 强类型化为您控制其类型的数据集，则可以很容易地使用 map 函数。您只需要定义一个案例类 + 编码器即可实现这一点。

我制作了一个非常简单的 csv 文件，其中包含以下确切内容（您的数据）：

Id;Subject;Marks
1;M,P,C;10,8,6
2;M,P,C;5,7,9
3;M,P,C;6,7,4

代码看起来像这样（大部分只是正确读取数据）：

import spark.implicits._
import org.apache.spark.sql.Encoders

case class MyClass(Id: Int, Subject: Seq[String], Marks: Seq[Int])
implicit val myEncoder = Encoders.product[MyClass]

val df = spark.read.option("header", "true").option("sep", ";").csv("csvWithArrays.csv")

val parsedDF = df
  .withColumn("Id", col("Id").cast("int"))
  .withColumn("Subject", split(col("Subject"), ","))
  .withColumn("Marks", split(col("Marks"), ",").cast("array<int>"))
  .as[MyClass]

val output = parsedDF.map{
  case MyClass(id, subject, marks) => {
    val maxMark = marks.max
    val maxSubject = subject(marks.indices.maxBy(marks))
    (id, maxSubject, maxMark) 
  }
}

output.show                                                                                                                                                                                                                                                              
+---+---+---+                                                                                                                                                                                                                                                                   
| _1| _2| _3|                                                                                                                                                                                                                                                                   
+---+---+---+                                                                                                                                                                                                                                                                   
|  1|  M| 10|                                                                                                                                                                                                                                                                   
|  2|  C|  9|                                                                                                                                                                                                                                                                   
|  3|  P|  7|                                                                                                                                                                                                                                                                   
+---+---+---+

一旦你对你的数据集进行了强类型化，你就可以轻松地使用 Scala 语言来获得你想要的东西:)

希望这可以帮助！