scala - significado - rdd spark

Cómo ordenar un RDD en Scala Spark?

Question

Método de lectura de Spark sortByKey:

sortByKey([ascending], [numTasks])   When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean ascending argument.

¿Es posible devolver solo una "N" cantidad de resultados? Entonces, en lugar de devolver todos los resultados, simplemente devuelva los 10 principales. Podría convertir la colección ordenada en una matriz y usar el método take , pero dado que se trata de una operación O (N), ¿hay un método más eficiente?

Answer 1

Lo más probable es que ya haya leído detenidamente el código fuente:

  class OrderedRDDFunctions {
   // <snip>
  def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.size): RDD[P] = {
    val part = new RangePartitioner(numPartitions, self, ascending)
    val shuffled = new ShuffledRDD[K, V, P](self, part)
    shuffled.mapPartitions(iter => {
      val buf = iter.toArray
      if (ascending) {
        buf.sortWith((x, y) => x._1 < y._1).iterator
      } else {
        buf.sortWith((x, y) => x._1 > y._1).iterator
      }
    }, preservesPartitioning = true)
  }

Y, como dices, la información completa debe pasar por la fase de reproducción aleatoria, como se ve en el fragmento.

Sin embargo, su preocupación acerca de la posterior invocación de la toma (K) puede no ser tan precisa. Esta operación NO recorre todos los N elementos:

  /**
   * Take the first num elements of the RDD. It works by first scanning one partition, and use the
   * results from that partition to estimate the number of additional partitions needed to satisfy
   * the limit.
   */
  def take(num: Int): Array[T] = {

Entonces, parece que:

O (myRdd.take (K)) << O (myRdd.sortByKey ()) ~ = O (myRdd.sortByKey.take (k)) (al menos para K pequeña) << O (myRdd.sortByKey (). Collect ()

Answer 2

Otra opción, al menos de PySpark 1.2.0, es el uso de takeOrdered .

En orden ascendente:

rdd.takeOrdered(10)

En orden descendente:

rdd.takeOrdered(10, lambda x: -x)

Los valores k superiores para los pares k, v:

rdd.takeOrdered(10, lambda (k, v): -v)

Answer 3

Si solo necesita los 10 primeros, use rdd.top(10) . Evita la clasificación, por lo que es más rápido.

rdd.top hace un pase paralelo a través de los datos, recolectando el N superior en cada partición en un montón, luego combina los montones. Es una operación O (rdd.count) . La ordenación sería O (rdd.count log rdd.count) e implicará una gran cantidad de transferencia de datos: realiza una reproducción aleatoria para que todos los datos se transmitan a través de la red.