python - 限制 Spark 上下文中的记录数量

Question

我想减少每个reducer的记录数量，并将结果变量保留为rdd

使用 takeSample 似乎是显而易见的选择，但是，它返回一个 集合 而不是 SparkContext 对象。

我想出了这个方法:

rdd = rdd.zipWithIndex().filter(lambda x:x[1]<limit).map(lambda x:x[0])

但是，这种方法非常慢且效率不高。

是否有更智能的方法来获取小样本并保持数据结构为rdd？

Answer 1

如果您想要一个小示例子集，并且无法对数据做出任何其他假设，那么take与parallelize相结合可能是最佳解决方案:

sc.parallelize(rdd.take(n))

它将涉及相对较少数量的分区(在最好的情况下只有一个)，并且小型 n 的网络流量成本应该可以忽略不计。

采样(randomSplit 或 sample)将需要与使用 filter 的 zipWithIndex 相同的完整数据扫描。

假设没有数据偏差，您可以尝试类似的方法来解决这个问题:

from __future__ import division  # Python 2 only

def limitApprox(rdd, n, timeout):
    count = rdd.countApprox(timeout)
    if count <= n:
        return rdd
    else:
        rec_per_part = count // rdd.getNumPartitions()
        required_parts = n / rec_per_part if rec_per_part else 1
        return rdd.mapPartitionsWithIndex(
            lambda i, iter: iter if i < required_parts else []
        )

• 这仍然会访问每个分区，但如果没有必要，会尝试避免计算内容
• 如果数据偏差较大，则不起作用
  • 如果分布均匀，但 n << 超过每个分区的平均记录数，则可能会超出所需的数量。
  • 如果分布偏向高指数，则可能会出现采样不足的情况。

如果数据可以表示为行，您可以尝试另一个技巧:

rdd.toDF().limit(n).rdd