spark transform系列__sample

Sample是对rdd中的数据集进行采样,并生成一个新的RDD,这个新的RDD只有原来RDD的部分数据,这个保留的数据集大小由fraction来进行控制,这个分析中,不分析sample的两个算法的具体实现,如果后期有必要时,可以分析这两个算法的具体的实现.

首先,先看看sample的实现代码:

def sample(
    withReplacement: Boolean,
    fraction: Double,
    seed: Long = Utils.random.nextLong): RDD[T] = withScope {
  require(fraction >= 0.0, "Negative fraction value: " + fraction)
  if (withReplacement) {
    new PartitionwiseSampledRDD[T, T](this, new PoissonSampler[T](fraction), true, 

       seed)
  } else {
    new PartitionwiseSampledRDD[T, T](this, new BernoulliSampler[T](fraction),

       true, seed)
  }
}

 

代码中的参数说明:

withReplacement=>,这个值如果是true时,采用PoissonSampler取样器(Poisson分布),

                      否则使用BernoulliSampler的取样器.

Fraction=>,一个大于0,小于或等于1的小数值,用于控制要读取的数据所占整个数据集的概率.

Seed=>,这个值如果没有传入,默认值是一个0~Long.maxvalue之间的整数.

 

流程说明:

在执行simple的transform时,会直接生成一个PartitionwiseSampledRDD的RDD实例,

生成PartitionwiseSampledRDD实例时,传入的第三个参数true表示Partitioner(partition的分区算子)使用上层依赖RDD的实例.

这个实例中的compute函数,针对RDD的上层依赖的iterator执行迭代时,会通过传入的sampler的实例,对数据进行取样操作,并返回取样的数据.

override def compute(splitIn: Partition, context: TaskContext): Iterator[U] = {
  val split = splitIn.asInstanceOf[PartitionwiseSampledRDDPartition]
  val thisSampler = sampler.clone

这里的seed根据传入的seed,每一个partition取出一个0~seed之间的一个随机的整数值(每个partition的seed的值有可能是不一样的,很大程度上,可能是不一样的.).
  thisSampler.setSeed(split.seed)

通过PoissonSampler或者BernoulliSampler的实例中的sample函数得到要返回的数据.

这里相当于是对上一个RDD中再嵌套了一层函数.
  thisSampler.sample(firstParent[T].iterator(split.prev, context))
}

 

针对每一个partition随机生成的seed的值的部分代码:

override def getPartitions: Array[Partition] = {
  val random = new Random(seed)
  firstParent[T].partitions.map(x => new PartitionwiseSampledRDDPartition(x, 

      random.nextLong()))
}