取数据超过内存限制的问题-解决方案（sample，takeSample，filter）

Posted 2021-02-03 wxplmm

遇到的问题

在处理数据过程中，遇到需要取(n)个数的问题，而当样本量过大的时候，就不能简单的take(n)，这类问题一般有两种情况：
> - 有序取 TopN
> - 无序取 N

---

先来讨论无序取N的情况：

• sample函数
  • sample(boolean, fraction,seed) : 按比例抽取

  • 返回一个新的RDD

    withReplacement：元素可以多次抽样(在抽样时替换)

    • withReplacement=true，表示有放回的抽样
    • withReplacement=false，表示无放回的抽样

    　
    fraction：期望样本的大小作为RDD大小的一部分， 当withReplacement=false时：选择每个元素的概率;分数一定是[0,1] ； 当withReplacement=true时：选择每个元素的期望次数; 分数必须大于等于0
    seed：随机数生成器的种子

图 11中 的 每 个 方 框 是 一 个 RDD 分 区。 通 过 sample 函 数， 采 样 50% 的 数 据。V1、 V2、 U1、 U2、U3、U4 采样出数据 V1 和 U1、 U2 形成新的 RDD

图  sample 算子对 RDD 转换

• takeSample函数
  • takeSample(boolean, sampleNum,seed) : 按固定数量抽取
  • 返回一个Array[T]； 该方法仅在预期结果数组很小的情况下使用，因为所有数据都被加载到driver的内存中
  • takeSample函数先是计算fraction，也就是采样比例，然后调用sample函数进行采样，并对采样后的数据进行collect()，最后调用take函数返回num个元素

withReplacement：元素可以多次抽样(在抽样时替换)

• withReplacement=true，表示有放回的抽样
• withReplacement=false，表示无放回的抽样

num：返回的样本的大小
seed：随机数生成器的种子

图  takeSample算子对RDD转换

---

再来看一下有序取 TopN的情况：

• filter函数
  • 函数功能是对元素进行过滤，对每个元素应用 f 函 数，返回值为 true 的元素在RDD中保留，返回值为 false 的元素将被过滤掉。 内 部 实 现 相 当 于 生 成 FilteredRDD(this，sc.clean(f))
  • 若是单列无法过滤，可以手动设置过滤位
  • 有点遗憾的是无法准确的取固定量的数

　图中每个方框代表一个 RDD 分区， T 可以是任意的类型。通过用户自定义的过滤函数 f，对每个数据项操作，将满足条件、返回结果为 true 的数据项保留。例如，过滤掉 V2 和 V3 保留了 V1，为区分命名为 V‘1。

　　图 filter 算子对 RDD 转换

---

附：takeSample源码

def takeSample(
    withReplacement: Boolean,
    num: Int,
    seed: Long = Utils.random.nextLong): Array[T] =
    {
        val numStDev = 10.0
        if (num < 0) {
            throw new IllegalArgumentException("Negative number of elements requested")
        } else if (num == 0) {
            return new Array[T](0)
        }
        val initialCount = this.count()
        if (initialCount == 0) {
            return new Array[T](0)
        }
        val maxSampleSize = Int.MaxValue - (numStDev * math.sqrt(Int.MaxValue)).toInt
        if (num > maxSampleSize) {
            throw new IllegalArgumentException("Cannot support a sample size > Int.MaxValue - "         + s"$numStDev * math.sqrt(Int.MaxValue)")
        }
        val rand = new Random(seed)
        if (!withReplacement && num >= initialCount) {
            return Utils.randomizeInPlace(this.collect(), rand)
        }
        val fraction = SamplingUtils.computeFractionForSampleSize(num, initialCount,        withReplacement)
        var samples = this.sample(withReplacement, fraction, rand.nextInt()).collect()
        // If the first sample didn‘t turn out large enough, keep trying to take samples;
        // this shouldn‘t happen often because we use a big multiplier for the initial size
        var numIters = 0
        while (samples.length < num) {
            logWarning(s"Needed to re-sample due to insufficient sample size. Repeat #$numIters")
            samples = this.sample(withReplacement, fraction, rand.nextInt()).collect()
            numIters += 1
        }
        Utils.randomizeInPlace(samples, rand).take(num)
}