Bagaimana untuk Memilih Baris Pertama Setiap Kumpulan dengan Nilai Tertinggi dalam Spark DataFrame dengan Cekap?

Cara memilih halaman utama setiap kumpulan

Matlamatnya adalah untuk mengekstrak halaman pertama dengan nilai tertinggi daripada setiap kumpulan DataFrame. Ini selalunya berlaku apabila menganalisis aliran data atau mengenal pasti pemain teratas dalam kategori tertentu. Untuk mencapai matlamat ini, beberapa kaedah boleh digunakan:

Fungsi tetingkap:

Fungsi tetingkap menyediakan cara untuk melakukan pengiraan dalam kumpulan. Dalam kes ini, kita boleh menggunakan fungsi row_number() untuk menetapkan nombor jujukan kepada setiap baris berdasarkan susunan yang ditentukan. Halaman utama kemudiannya boleh dikenal pasti dengan menapis kedudukan kepada 1.

import org.apache.spark.sql.functions.{row_number, max, broadcast}
import org.apache.spark.sql.expressions.Window

val df = sc.parallelize(Seq(
  (0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3),
  (1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3),
  (2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8),
  (3,"cat8",35.6))).toDF("Hour", "Category", "TotalValue")

val w = Window.partitionBy($"hour").orderBy($"TotalValue".desc)

val dfTop = df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Caburan pasca pengagregatan SQL mudah:

Sebagai alternatif, kita boleh menggunakan SQL untuk melaksanakan pengagregatan dan kemudian menggabungkan hasil dengan DataFrame asal untuk mengekstrak halaman pertama setiap kumpulan.

val dfMax = df.groupBy($"hour".as("max_hour")).agg(max($"TotalValue").as("max_value"))

val dfTopByJoin = df.join(broadcast(dfMax),
    ($"hour" === $"max_hour") && ($"TotalValue" === $"max_value"))
  .drop("max_hour")
  .drop("max_value")

dfTopByJoin.show

// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Gunakan pengisihan struktur:

Cara yang kemas untuk mendapatkan hasil yang sama tanpa menggunakan fungsi tetingkap atau cantuman ialah mengisih data berdasarkan struktur yang mengandungi nilai dan kategori. Nilai maksimum struktur ini kemudiannya akan mengembalikan halaman pertama yang dikehendaki untuk setiap kumpulan.

val dfTop = df.select($"Hour", struct($"TotalValue", $"Category").alias("vs"))
  .groupBy($"hour")
  .agg(max("vs").alias("vs"))
  .select($"Hour", $"vs.Category", $"vs.TotalValue")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+

Atas ialah kandungan terperinci Bagaimana untuk Memilih Baris Pertama Setiap Kumpulan dengan Nilai Tertinggi dalam Spark DataFrame dengan Cekap?. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!