Apache spark：对的示例 RDD答案

【问题标题】：Apache spark: sample RDD of pairsApache spark：对的示例 RDD
【发布时间】：2016-12-18 13:36:53
【问题描述】：

我有一个项目的 RDD，以及一个计算两个项目之间距离的函数 d: (Item, Item) => Double。我正在尝试计算从 RDD 中随机抽取的项目之间的平均距离。 RDD 相当大（数以百万计），因此计算确切的平均值是不可能的。

因此，我想获得采样项目对的 RDD（我将从中计算距离）。例如，我想获得 100m 对的样本。给定采样对的 RDD，然后我将计算平均值、直方图等以了解距离分布。

以下是所有失败的初始尝试：

使用.sample 生成两个RDD，压缩它们并计算项目之间的距离。这失败了，因为.zip 要求两个 RDD 每个分区的项目数完全相同。
将RDD的.cartesian与自身一起使用，然后使用.sample。这会失败（内存不足），因为显然cartesian 不应该以这种方式使用。
收集 RDD 的两个小样本，并 .zip 两个数组。这工作正常，但无法扩展。

有什么想法吗？

谢谢！

编辑：这里是如何压缩每个分区具有不同数量的项目的两个样本：

val r = ... // RDD[Item]
val s1 = r.sample(true, 0.1, 123)
val s2 = r.sample(true, 0.1, 456)
val zipper = (i1: Iterator[Item], i2: Iterator[Item]) => i1.zip(i2)
val pairs = r1.zipPartitions(r2)(zipper) // zip the RDDs and explicitly define how to zip the partitions

关键是，虽然 RDD 的 .zip 方法不接受大小不等的分区，但迭代器的 .zip 方法可以（并丢弃较长迭代器的剩余部分）。

【问题讨论】：

“这失败了（内存不足），因为笛卡尔显然不应该以这种方式使用。”你能详细说明为什么你认为它不应该被使用那样吗？
你试过rdd.cartesian(rdd).filter(lambda (a, b): a.key < b.key). ...这样的方法吗？这样可以减少一半的数据
@YuvalItzchakov .cartesian 的问题是 spark 尝试创建 n^2 个分区，每个分区大约有 m^2 个元素（假设最初有 n 个分区，每个分区大约 m 个项目）。问题是，我不需要那么多分区（或那么多项目对）。典型的数字是：50 个分区，每个分区有 10M 项。即使存储笛卡尔积 RDD 的单个分区也是不可行的。我试图以 1e-7 的因子对其进行采样，但没有成功。我猜是因为 spark 实际上在执行器上创建了产品分区。
@AlbertoBonsanto 感谢您的评论。但我的问题不是 2 的因数，而是更接近 1e6 的因数……我不需要实际的 .cartesian RDD，而只需要它的一小部分。如上所述，.sample 不起作用。
@Amir 好的，这就是我想知道的。根据您所说，我认为 API 存在问题，但我理解 n ^ 2 问题，您实际上并不需要那么多样本。

标签： apache-spark random rdd

【解决方案1】：

回答我自己的问题：

获取 rdd 的样本（带替换），
使用.sliding(2) 获取样本的连续对。

代码：

import org.apache.spark.mllib.rdd.RDDFunctions._ // for .sliding
val x = ... // RDD[Item]
val xSize = x.count
val n = 1000000.0 // (approximate) desired sample size
val pairs = x.sample(true, n/xSize).sliding(2)
val distances = pairs.map(arr => dist(arr(0), arr(1)))

【讨论】：

注意：这并不完全是随机绘制的项目对（因为，例如，每个元素都保证至少出现两次）。一个准确的解决方案是使用等效于 scala 的.grouped，但对于 RDD（我找不到）。当然，这可以使用.mapPartitions 显式编写，但.sliding 解决方案对我来说已经足够了。