विभाजित और कुशलता से समूह <code>RDD[K, V]</code> के लिए एक समाधान RDD समूहों की गणना के अनुसार मुख्य

Question

कुंजी के द्वारा मैं क्रियान्वित किया है और प्रत्येक समूह (K, RDD[V]), partitionBy और Partitioner का उपयोग कर के अनुसार डेटा की गणना करने के लिए PartitionBy का उपयोग करना। फिर भी, मुझे यकीन नहीं है कि यह वास्तव में कुशल है और मैं आपका दृष्टिकोण देखना चाहता हूं।

यहां नमूने मामला है: [K: Int, V: Int] की एक सूची के अनुसार, गणना V रों K के प्रत्येक समूह के लिए मतलब है, जानते हुए भी कि यह वितरित किया जाना चाहिए और V मूल्यों बहुत बड़ी हो सकती है। यही कारण है कि देना चाहिए:

List[K, V] => (K, mean(V)) 

सरल विभाजनर वर्ग:

class MyPartitioner(maxKey: Int) extends Partitioner { 

    def numPartitions = maxKey 

    def getPartition(key: Any): Int = key match { 
     case i: Int if i < maxKey => i 
    } 
    } 

विभाजन कोड:

val l = List((1, 1), (1, 8), (1, 30), (2, 4), (2, 5), (3, 7)) 

     val rdd = sc.parallelize(l) 
     val p = rdd.partitionBy(new MyPartitioner(4)).cache() 

     p.foreachPartition(x => { 
     try { 
      val r = sc.parallelize(x.toList) 
      val id = r.first() //get the K partition id 
      val v = r.map(x => x._2) 
      println(id._1 + "->" + mean(v)) 
     } catch { 
      case e: UnsupportedOperationException => 0 
     } 
     }) 

उत्पादन होता है:

1->13, 2->4, 3->7

मेरे प्रश्न हैं:

1. partitionBy पर कॉल करते समय वास्तव में क्या होता है? (क्षमा करें, मुझे उस पर पर्याप्त चश्मा नहीं मिला)
2. क्या यह विभाजन द्वारा मैप करना वास्तव में कुशल है, यह जानकर कि मेरे उत्पादन मामले में यह बहुत अधिक मूल्यों (नमूना के लिए 50 के रूप में) बहुत अधिक मूल्यों से नहीं होगा (जैसा कि नमूना के लिए 1 मिलियन)
3. paralellize(x.toList) की लागत क्या है? क्या यह करने के लिए लगातार है? (मुझे mean() के इनपुट में RDD की आवश्यकता है)
4. आप इसे अपने आप कैसे करेंगे?

सादर

Answer 1

आपका कोड काम नहीं करना चाहिए। आप निष्पादकों को SparkContext ऑब्जेक्ट पास नहीं कर सकते हैं। (यह Serializable नहीं है।) इसके अलावा मुझे नहीं पता कि आपको क्यों आवश्यकता होगी।

माध्य की गणना करने के लिए, आपको योग और गणना की गणना करने और अपना अनुपात लेने की आवश्यकता है। डिफ़ॉल्ट विभाजनकर्ता ठीक करेगा।

def meanByKey(rdd: RDD[(Int, Int)]): RDD[(Int, Double)] = { 
    case class SumCount(sum: Double, count: Double) 
    val sumCounts = rdd.aggregateByKey(SumCount(0.0, 0.0))(
    (sc, v) => SumCount(sc.sum + v, sc.count + 1.0), 
    (sc1, sc2) => SumCount(sc1.sum + sc2.sum, sc1.count + sc2.count)) 
    sumCounts.map(sc => sc.sum/sc.count) 
} 

यह एक कुशल एकल-पास गणना है जो अच्छी तरह से सामान्यीकृत होती है।