¿Cómo selecciono un rango de elementos en Spark RDD?

No creo que haya un método eficiente para hacer esto todavía. Pero la manera fácil es usar filter(), digamos que tienes un RDD, pairs con pares de valores clave y solo quieres elementos de 60 a 80 inclusive solo hazlo.

val 60to80 = pairs.filter {
    _ match {
        case (k,v) => k >= 60 && k <= 80
        case _ => false //incase of invalid input
    }
}

Creo que es posible que esto se pueda hacer de manera más eficiente en el futuro, utilizando sortByKey y guardando información sobre el rango de valores asignados a cada partición. Tenga en cuenta que este enfoque solo guardaría algo si estuviera planeando consultar el rango varias veces porque el tipo es obviamente caro.

Mirando la fuente spark definitivamente sería posible hacer consultas de rango eficientes usando RangePartitioner:

// An array of upper bounds for the first (partitions - 1) partitions
  private val rangeBounds: Array[K] = {

Este es un miembro privado de RangePartitioner con el conocimiento de todos los límites superiores de las particiones, sería fácil consultar solo las particiones necesarias. Parece que esto es algo que los usuarios de spark pueden ver en el futuro: SPARK-911

ACTUALIZACIÓN: Respuesta mucho mejor, basada en pull solicitud que estoy escribiendo para SPARK-911. Se ejecutará de manera eficiente si el RDD está ordenado y lo consulta varias veces.

val sorted = sc.parallelize((1 to 100).map(x => (x, x))).sortByKey().cache()
val p: RangePartitioner[Int, Int] = sorted.partitioner.get.asInstanceOf[RangePartitioner[Int, Int]];
val (lower, upper) = (10, 20)
val range = p.getPartition(lower) to p.getPartition(upper)
println(range)
val rangeFilter = (i: Int, iter: Iterator[(Int, Int)]) => {
  if (range.contains(i))
    for ((k, v) <- iter if k >= lower && k <= upper) yield (k, v)
  else
    Iterator.empty
}
for((k,v) <- sorted.mapPartitionsWithIndex(rangeFilter, preservesPartitioning = true).collect()) println(s"$k, $v")

Si tener toda la partición en memoria es aceptable, incluso podría hacer algo como esto.
val glommedAndCached = sorted.glom()cache(); glommedAndCached.map(a => a.slice(a.search(lower),a.search(upper)+1)).collect()

search no es un miembro Por cierto acabo de hacer una clase implícita que tiene una función de búsqueda binaria, no se muestra aquí