我想在 Spark RDD 中选择一系列元素。比如我有一个有一百个元素的RDD，我需要选择60到80个元素，我该怎么做呢？
我看到 RDD 有一个 take(i: int) 方法，它返回前 i 个元素。但是没有对应的方法来取最后的i个元素，或者从某个索引开始从中间开始的i个元素。
【问题讨论】：
ROW_NUMBER() 或 RANK() 添加到数据集，然后 @ 987654324@ 所需的行。对于较小的 RDD，这是多余的，但这种方法应该对非常大的 RDD 有效。
标签： apache-spark rdd
对于那些偶然发现这个问题并寻找与 Spark 2.x 兼容的答案的人，您可以使用filterByRange
【讨论】：
我认为目前还没有有效的方法来做到这一点。但最简单的方法是使用filter()，假设你有一个RDD，pairs 和键值对，你只想要60 到80 之间的元素就行了。
val 60to80 = pairs.filter {
_ match {
case (k,v) => k >= 60 && k <= 80
case _ => false //incase of invalid input
}
}
我认为通过使用sortByKey 并保存有关映射到每个分区的值范围的信息，将来可能会更有效地完成此操作。请记住，如果您计划多次查询范围，这种方法只会节省任何东西，因为排序显然很昂贵。
通过查看 spark 源，肯定可以使用 RangePartitioner 进行有效的范围查询：
// An array of upper bounds for the first (partitions - 1) partitions
private val rangeBounds: Array[K] = {
这是RangePartitioner 的私有成员，知道所有分区的上限，因此很容易只查询必要的分区。看起来这是火花用户将来可能会看到的东西：SPARK-911
更新：更好的答案，基于我为 SPARK-911 编写的拉取请求。如果对 RDD 进行排序并且您多次查询它，它将有效地运行。
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")
如果将整个分区放在内存中是可以接受的，您甚至可以这样做。val glommedAndCached = sorted.glom()cache();
glommedAndCached.map(a => a.slice(a.search(lower),a.search(upper)+1)).collect()
search 不是成员 BTW 我只是做了一个具有二分查找功能的隐式类，此处未显示
【讨论】：
case (k,v) => k >= 60 || k <= 80 没有&& 而不是||？
以下应该可以得到范围。注意缓存会为你节省一些开销，因为 zipWithIndex 内部需要扫描 RDD 分区以获取每个分区中的元素数。
scala>val r1 = sc.parallelize(List("a", "b", "c", "d", "e", "f", "g"), 3).cache
scala>val r2 = r1.zipWithIndex
scala>val r3 = r2.filter(x=> {x._2>2 && x._2 < 4}).map(x=>x._1)
scala>r3.foreach(println)
d
【讨论】：
您的数据集有多大？您也许可以做您需要的事情：
data.take(80).drop(59)
这似乎效率低下，但对于中小型数据，应该可以。
是否可以通过其他方式解决此问题？从数据中间准确挑选某个范围的情况是什么？ takeSample 会更好地为您服务吗？
【讨论】：
takeSample 也不起作用，我需要一个范围（从...到）。
getRange (from,to) 在分布式模型中不起作用。
take() 是一个动作，它将强制在驱动程序处收集 RDD。它确实只适用于适合驾驶员记忆的小型设备。如果您没有索引或排序，range(...) 是不稳定的，所以我想您需要某种形式的键来进行选择。