过滤和合并时避免重新分配成本

Question

我正在 pyspark 中对 （x，y） 点的 RDD 实现范围查询。我将 xy 空间划分为一个 16*16 的网格（256 个单元格），并将 RDD 中的每个点分配给其中一个单元格。gridMappedRDD 是一个 PairRDD：（cell_id，Point 对象）

我将这个RDD分区为256个分区，使用：

gridMappedRDD.partitionBy(256)

范围查询是一个矩形框。我为我的网格对象准备了一个方法，它可以返回与查询范围重叠的单元格id列表。所以，我用这个作为过滤器来删除不相关的单元格:

filteredRDD = gridMappedRDD.filter(lambda x: x[0] in candidateCells)

但问题是，在运行查询然后收集结果时，会评估所有 256 个分区;为每个分区创建一个任务。

为了避免这个问题，我尝试将过滤后的RDD合并为候选单元格列表的长度，我希望这可以解决问题。

filteredRDD.coalesce(len(candidateCells))

事实上，生成的RDD具有< code>len(candidateCells)分区，但这些分区与< code>gridMappedRDD不同。

正如在coalesce留档中所述，schffle参数是False，不应该在分区之间执行随机切换，但我可以看到（在Glom（）的帮助下）情况并非如此。

例如，在与candidateCells=[62，63，78，79]合并（4）之后，分区如下：

[[(62, P), (62, P) .... , (63, P)],
 [(78, P), (78, P) .... , (79, P)],
 [], []
]

实际上，通过合并，我有一个随机读取，它等于我的每个任务的整个数据集的大小，这需要很长时间。我需要的是一个RDD，只有与candidateCells中的单元格相关的分区，没有任何洗牌。所以，我的问题是，有没有可能只过滤一些分区而不重新洗牌？对于上面的例子，我的filteredRDD将有4个分区，其数据与originalRDD的62、63、78、79个分区完全相同。这样做，查询可以只针对受影响的分区。

Answer 1

您在这里做出了一些不正确的假设：

• 洗牌与合并无关（合并在这里也没有用）。它是由分区引起的。根据定义，分区需要随机播放。
• 分区不能用于优化筛选器。Spark对你使用的功能一无所知（它是一个黑匣子）。
分区
• 不会唯一地将键映射到分区。多个键可以放在同一个分区上 - HashPartitioner 如何工作？

你能做什么:

>

如果结果子集是小的重新分区，并对每个键应用查找：

from itertools import chain

partitionedRDD = gridMappedRDD.partitionBy(256)

chain.from_iterable(
    ((c, x) for x in partitionedRDD.lookup(c)) 
    for c in candidateCells
)

如果数据很大，您可以尝试跳过扫描分区（任务数量不会改变，但某些任务可能会短路）：

candidatePartitions = [
    partitionedRDD.partitioner.partitionFunc(c) for c in candidateCells
]

partitionedRDD.mapPartitionsWithIndex(
    lambda i, xs: (x for x in xs if x[0] in candidateCells) if i in candidatePartitions else []
)

这两种方法只有在执行多次“查找”时才有意义。如果是一次性操作，最好执行线性滤波：

• 它比随机和重新分区便宜。
• 如果初始数据均匀分布，则下游处理将能够更好地利用可用资源。