Является ли topByKey масштабируемым?

Question

Когда я использую topByKey с разными кластерами, у меня одинаковое количество времени, чтобы выполнить этот код независимо от количества используемых slave-устройств. Размер RDD_distance составляет 10^8 и 10^12 единиц.

 parsedData.cache 

     for(ind <- 1 to maxIterForYstar ) { 

     var rdd_distance = rdd_temp.cartesian(parsedData).map{ case (x,y) => (x.get_id,(y.get_vector,-Vectors.sqdist(x.get_vector,y.get_vector))) } 

     var rdd_knn_bykey = rdd_distance.topByKey(k)(Ordering[(Double)].on(x=>x._2)) 
    } 

Итак, мой вопрос в том, чтобы знать, является ли topByKey масштабируемым или если что-то не так с моим кодом.

Answer 1

Вот фактический код для topByKey:

@DeveloperApi 
class MLPairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) extends Serializable { 
    /** 
    * Returns the top k (largest) elements for each key from this RDD as defined by the specified 
    * implicit Ordering[T]. 
    * If the number of elements for a certain key is less than k, all of them will be returned. 
    * 
    * @param num k, the number of top elements to return 
    * @param ord the implicit ordering for T 
    * @return an RDD that contains the top k values for each key 
    */ 
    def topByKey(num: Int)(implicit ord: Ordering[V]): RDD[(K, Array[V])] = { 
    self.aggregateByKey(new BoundedPriorityQueue[V](num)(ord))(
     seqOp = (queue, item) => { 
     queue += item 
     }, 
     combOp = (queue1, queue2) => { 
     queue1 ++= queue2 
     } 
    ).mapValues(_.toArray.sorted(ord.reverse)) // This is an min-heap, so we reverse the order. 
    } 
} 

Так что это единственный способ в MLPairRDDFunctions. Это более эффективный groupByKey, только верхние k элементов для каждой сохраненной клавиши.

несколько вопросов, чтобы рассмотреть, что вы можете ответить, изучая Спарк веб-консоль:

• Сколько параллельных задач на самом деле выполняется? Некоторые из ваших дополнительных рабов могут не использоваться.
• Где же действительно расходуется время? Например, действительно ли самый дешевый продукт cartesian? Не могли бы вы использовать кеширование более эффективно? Кроме того, похоже, что rdd_distance может быть вычислен один раз, до выражения for, или это просто упрощенный пример?