स्पार्क डेटाफ्रेम में Argmax: अधिकतम मूल्य

Question

स्पार्क डेटाफ्रेम df को देखते हुए, मैं एक निश्चित संख्यात्मक कॉलम 'values' में अधिकतम मान प्राप्त करना चाहता हूं, और उस मान को प्राप्त करें जहां वह मान प्राप्त हुआ था। मैं निश्चित रूप से यह कर सकते हैं:

# 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() 

लेकिन यह अक्षम है, क्योंकि यह df के माध्यम से दो गुजरता की आवश्यकता है।

pandas.Series/DataFrame और numpy.arrayargmax/idxmax तरीकों कि इस कुशलता से करने (एक पास में) है। तो मानक पायथन (अंतर्निहित फ़ंक्शन max एक प्रमुख पैरामीटर स्वीकार करता है, इसलिए इसका उपयोग उच्चतम मान की अनुक्रमणिका को खोजने के लिए किया जा सकता है)।

स्पार्क में सही दृष्टिकोण क्या है? ध्यान दें कि मुझे कोई फर्क नहीं पड़ता कि मुझे उन सभी पंक्तियां मिलती हैं जहां अधिकतम मूल्य प्राप्त होता है, या उन पंक्तियों के कुछ मनमाने ढंग से (खाली नहीं!) सबसेट।

Answer 1

स्कीमा is Orderable (स्कीमा केवल एटोमिक्स/एटोमिक्स के एरे/रिकर्सिवली orderable structs होता है) यदि आप सरल एकत्रित उपयोग कर सकते हैं:

अजगर:

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

स्काला:

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

नहीं तो आप कम कर सकते हैं Dataset से अधिक (स्काला केवल), लेकिन यह अतिरिक्त अक्रमांकन की आवश्यकता है:

type T = ??? 

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

Answer 2

शायद यह एक अधूरा जवाब है, लेकिन आप का उपयोग करें 'DataFrame आंतरिक RDD कर सकते हैं, max विधि लागू करते हैं और एक निर्धारित कुंजी का उपयोग कर अधिक से अधिक रिकॉर्ड मिलता है।

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)