argmax: satırın maksimum değeriyle nasıl alınacağı

Question

Bir Spark DataFrame df verildiğinde, belirli bir sayısal sütun 'values''daki maksimum değeri bulmak ve bu değere ulaşıldığı satırı (satırları) almak istiyorum. Tabii bunu yapabilirsiniz:

# it doesn't matter if I use scala or python, 
# since I hope I get this done with DataFrame API 
import pyspark.sql.functions as F 
max_value = df.select(F.max('values')).collect()[0][0] 
df.filter(df.values == max_value).show() 

ama df iki geçiş gerektirir çünkü bu verimsizdir.

pandas.Series/DataFrame ve numpy.array (bir geçişte) verimli bir şekilde yapmak argmax/idxmax yöntemleri vardır. Standart python (dahili fonksiyon max bir anahtar parametresini kabul eder, bu yüzden en yüksek değerin indeksini bulmak için kullanılabilir).

Spark'de doğru yaklaşım nedir? Maksimum değerin elde edildiği tüm satırları veya bu satırların yalnızca bazı (boş olmayan!) Alt kümelerini alıp almamaya dikkat etmem.

Answer 1

şema is Orderable Eğer basit toplamlar kullanabilirsiniz (şema sadece atomics/atomics ait diziler/yinelemeli orderable yapılar içerir) ise:

Python:

df.select(F.max(
    F.struct("values", *(x for x in df.columns if x != "values")) 
)).first() 

Scala:

df.select(max(struct(
    $"values" +: df.columns.collect {case x if x!= "values" => col(x)}: _* 
))).first 

Aksi takdirde azaltabilir Dataset üzerinde (Scala için) ama ek deserialization gerektirir: Genel olarak

type T = ??? 

df.reduce((a, b) => if (a.getAs[T]("values") > b.getAs[T]("values")) a else b) 

Answer 2

Belki de tamamlanmamış bir yanıttır, ancak DataFrame iç RDD'u kullanabilir, max yöntemini uygulayabilir ve belirlenen bir anahtarı kullanarak maksimum kaydı alabilirsiniz.

a = sc.parallelize([ 
    ("a", 1, 100), 
    ("b", 2, 120), 
    ("c", 10, 1000), 
    ("d", 14, 1000) 
    ]).toDF(["name", "id", "salary"]) 

a.rdd.max(key=lambda x: x["salary"]) # Row(name=u'c', id=10, salary=1000)