स्पार्क डेटाफ्रेम: समूहों पर संचालित

Question

मुझे एक डेटाफ्रेम मिला है जिस पर मैं काम कर रहा हूं, और मैं कॉलम के एक सेट द्वारा समूह करना चाहता हूं और शेष कॉलम पर प्रति-समूह संचालित करना चाहता हूं। नियमित RDD -land में मुझे लगता है कि यह कुछ इस तरह दिखेगा:

rdd.map(tup => ((tup._1, tup._2, tup._3), tup)). 
    groupByKey(). 
    forEachPartition(iter => doSomeJob(iter)) 

DataFrame में -land मैं इस तरह शुरू कर दूं:

df.groupBy("col1", "col2", "col3") // Reference by name 

लेकिन तब मुझे यकीन है कि पर काम करने के लिए कैसे नहीं कर रहा हूँ समूह यदि मेरे ऑपरेशन GroupedData द्वारा प्रस्तावित औसत/न्यूनतम/अधिकतम/गिनती से अधिक जटिल हैं।

उदाहरण के लिए, मैं (समूह में जुड़े रों के माध्यम से पुनरावृत्ति द्वारा) ("col1", "col2", "col3") समूह के लिए एक एकल MongoDB दस्तावेज़ निर्माण, N विभाजन करने के लिए नीचे पैमाने पर है, तो एक MongoDB डेटाबेस में डॉक्स सम्मिलित करना चाहते हैं। N सीमा एक साथ कनेक्शन के अधिकतम संख्या है जो मैं चाहता हूं।

कोई सलाह?

Answer 1

आप स्वयं से जुड़ सकते हैं। पहले समूह मिलती है:

val groups = df.groupBy($"col1", $"col2", $"col3").agg($"col1", $"col2", $"col3") 

तो फिर तुम मूल DataFrame को यह वापस शामिल हो सकते हैं: यह आप ठीक उसी डेटा आप मूल रूप से किया था हो जाता है जबकि

val joinedDF = groups 
    .select($"col1" as "l_col1", $"col2" as "l_col2", $"col3" as "l_col3) 
    .join(df, $"col1" <=> $"l_col1" and $"col2" <=> $"l_col2" and $"col3" <=> $"l_col3") 

(और 3 अतिरिक्त, अनावश्यक कॉलम के साथ) आप पंक्ति के साथ जुड़े (col1, col2, col3) समूह के लिए MongoDB दस्तावेज़ आईडी के साथ एक कॉलम जोड़ने के लिए एक और शामिल हो सकते हैं।

किसी भी दर पर, मेरे अनुभव में शामिल होने और स्वयं-जुड़ने के तरीके डेटाफ्रेम में जटिल सामग्री को संभालने के तरीके हैं।