PySpark में सॉर्ट किए गए कम करने का सबसे प्रभावी तरीका क्या है?

Question

मैं 2015 से अमेरिकी घरेलू उड़ानों के समय-समय पर प्रदर्शन रिकॉर्ड का विश्लेषण कर रहा हूं। मुझे पूंछ संख्या से समूह करने की आवश्यकता है, और मेरे आवेदन द्वारा पुनर्प्राप्त करने के लिए डेटाबेस में प्रत्येक पूंछ संख्या के लिए सभी उड़ानों की तिथि क्रमबद्ध सूची संग्रहित करने की आवश्यकता है। मुझे यकीन नहीं है कि इसे प्राप्त करने के लिए कौन से दो विकल्प सबसे अच्छे हैं।

# Load the parquet file 
on_time_dataframe = sqlContext.read.parquet('../data/on_time_performance.parquet') 

# Filter down to the fields we need to identify and link to a flight 
flights = on_time_dataframe.rdd.map(lambda x: 
    (x.Carrier, x.FlightDate, x.FlightNum, x.Origin, x.Dest, x.TailNum) 
) 

मैं एक तरह से कम करने में इस लक्ष्य को हासिल कर सकते हैं ...

# Group flights by tail number, sorted by date, then flight number, then 
origin/dest 
flights_per_airplane = flights\ 
    .map(lambda nameTuple: (nameTuple[5], [nameTuple]))\ 
    .reduceByKey(lambda a, b: sorted(a + b, key=lambda x: (x[1],x[2],x[3],x[4]))) 

या मैं बाद में एक नक्शा नौकरी में इसे प्राप्त कर सकते हैं ...

# Do same in a map step, more efficient or does pySpark know how to optimize the above? 
flights_per_airplane = flights\ 
    .map(lambda nameTuple: (nameTuple[5], [nameTuple]))\ 
    .reduceByKey(lambda a, b: a + b)\ 
    .map(lambda tuple: 
    (
     tuple[0], sorted(tuple[1], key=lambda x: (x[1],x[2],x[3],x[4]))) 
    ) 

को कम में ऐसा करने से वास्तव में अक्षम लगता है, लेकिन वास्तव में दोनों बहुत धीमी हैं। क्रमबद्ध() PySpark दस्तावेज़ों में ऐसा करने के तरीके की तरह दिखता है, तो मुझे आश्चर्य है कि क्या PySpark आंतरिक रूप से यह कोशेर नहीं बना रहा है? किसी अन्य कारण के लिए कौन सा विकल्प सबसे कुशल या सर्वोत्तम विकल्प है?

मेरे कोड एक सार यहाँ में भी है: https://gist.github.com/rjurney/af27f70c76dc6c6ae05c465271331ade

आप डेटा के बारे में उत्सुक हैं, तो यह परिवहन सांख्यिकी ब्यूरो, यहां से है: http://www.transtats.bts.gov/DL_SelectFields.asp?Table_ID=236&DB_Short_Name=On-Time

Answer 1

दुर्भाग्य से दोनों तरीकों से आप से पहले गलत हैं यहां तक ​​कि सॉर्टिंग शुरू करें और स्पार्क में ऐसा करने का कोई प्रभावी और सरल तरीका नहीं है। फिर भी, पहला एक दूसरे की तुलना में काफी खराब है।

दोनों तरीकों गलत क्यों हैं? क्योंकि यह सिर्फ एक और groupByKey है और यह केवल एक महंगा ऑपरेशन है। कुछ तरीकों से आप चीजों को बेहतर बनाने की कोशिश कर सकते हैं (विशेष रूप से मानचित्र पक्ष में कमी से बचने के लिए) लेकिन दिन के अंत में आपको केवल एक पूर्ण शफल की कीमत चुकानी पड़ेगी और यदि आपको कोई असफलता दिखाई नहीं दे रही है तो यह शायद लायक नहीं है सभी झगड़ा

फिर भी, दूसरा दृष्टिकोण एल्गोरिदमिक रूप से बेहतर है *। यदि आप पहले प्रयास में तरह से क्रमबद्ध संरचना को रखना चाहते हैं तो आपको समर्पित उपकरण (aggregateByKeybisect.insort के साथ एक अच्छी पसंद होगी) लेकिन यहां हासिल करने के लिए वास्तव में कुछ भी नहीं है।

यदि समूहित आउटपुट एक कठिन आवश्यकता है तो आप सबसे अच्छी चीज कर सकते हैं keyBy, groupByKey और सॉर्ट करें। यह दूसरा समाधान के साथ प्रदर्शन में सुधार नहीं होगा लेकिन यकीनन पठनीयता में सुधार होगा:

(flights 
    .keyBy(lambda x: x[5]) 
    .groupByKey() 
    .mapValues(lambda vs: sorted(vs, key=lambda x: x[1:5]))) 

---

* यहां तक ​​कि अगर आप यह मान Timsort पहले दृष्टिकोण के लिए सबसे अच्छा स्थिति N बार हे (एन) जबकि है दूसरा सबसे खराब स्थिति परिदृश्य में ओ (एन लॉग एन) है।