Как получить минимальное и максимальное значения столбцов?

Question

Я хочу провести концептуальную проверку своего кода. Цель состоит в том, чтобы вычислить минимальное значение поля minTimestamp и максимальное значение поля maxTimestamp в кадре данных df и удалить все остальные значения. Например:

ДФ

src    dst  minTimestamp   maxTimestamp
1      3    1530809948     1530969948
1      3    1540711155     1530809945
1      3    1520005712     1530809940
2      3    1520005712     1530809940

Ответ должен быть следующим:

результат:

src    dst  minTimestamp   maxTimestamp
1      3    1520005712     1530969948
2      3    1520005712     1530809940

Это мой код:

val cw_min = Window.partitionBy($"src", $"dst").orderBy($"minTimestamp".asc)
val cw_max = Window.partitionBy($"src", $"dst").orderBy($"maxTimestamp".desc)

val result = df
  .withColumn("rn", row_number.over(cw_min)).where($"rn" === 1).drop("rn")
  .withColumn("rn", row_number.over(cw_max)).where($"rn" === 1).drop("rn")

Можно ли использовать функцию Window последовательно, как я делал в моем примере кода? Проблема в том, что я всегда получаю одинаковые значения minTimestamp и maxTimestamp.

Answer 1

Вы можете использовать DataFrame groupBy для агрегирования min и max:

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

val df = Seq(
  (1, 3, 1530809948L, 1530969948L),
  (1, 3, 1540711155L, 1530809945L),
  (1, 3, 1520005712L, 1530809940L),
  (2, 3, 1520005712L, 1530809940L)
).toDF("src", "dst", "minTimestamp", "maxTimestamp")

df.groupBy("src", "dst").agg(
    min($"minTimestamp").as("minTimestamp"), max($"maxTimestamp").as("maxTimestamp")
  ).
  show
// +---+---+------------+------------+                                             
// |src|dst|minTimestamp|maxTimestamp|
// +---+---+------------+------------+
// |  2|  3|  1520005712|  1530809940|
// |  1|  3|  1520005712|  1530969948|
// +---+---+------------+------------+

Answer 2

Почему бы не использовать Spark SQL и не сделать

val spark: SparkSession = ???
df.createOrReplaceTempView("myDf")
val df2 = spark.sql("""
    select 
      src, 
      dst, 
      min(minTimestamp) as minTimestamp, 
      max(maxTimestamp) as maxTimestamp 
    from myDf group by src, dst""")

Вы также можете использовать API, чтобы сделать то же самое:

val df2 = df
   .groupBy("src", "dst")
   .agg(min("minTimestamp"), max("maxTimestamp"))