为什么过滤器不保留分区？答案

【问题标题】：Why filter does not preserve partitioning?为什么过滤器不保留分区？
【发布时间】：2018-05-11 08:48:23
【问题描述】：

一些操作，例如map、flatMap、filter，不保留分区。 map、flatMap、filter 操作对每个分区应用一个函数。

我不明白为什么过滤器不保留分区。它只是获取满足条件的每个分区的子集，因此我认为可以保留分区。为什么不是这样？

【问题讨论】：

【解决方案1】：

你当然是对的。报价是不正确的。 filter 确实保留了分区（出于您已经描述的原因），确认这一点很简单

val rdd = sc.range(0, 10).map(x => (x % 3, None)).partitionBy(
  new org.apache.spark.HashPartitioner(11)
)

rdd.partitioner
// Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@b)

val filteredRDD = rdd.filter(_._1 == 3)
filteredRDD.partitioner
// Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@b)

rdd.partitioner == filteredRDD.partitioner
// Boolean = true

这与 map 之类的操作形成对比，后者不保留分区 (Partitioner)：

rdd.map(identity _).partitioner
// Option[org.apache.spark.Partitioner] = None

Datasets 有点微妙，因为过滤器通常是下推的，但总体而言行为相似。

【讨论】：

我不太明白你的推理，即使rdds具有相同的分区器，这是否意味着分区是相同的？它甚至意味着“保留分区”是什么意思？对我来说，保留分区意味着分区的数量保持不变，并且分区之间不会发生交换（洗牌）。但这也适用于map
@RaphaelRoth Default Partitioning Scheme in Spark。 preservesPartitioning 意味着保留Partitioner 而不是保留分区数。如果你 map 产生 RDD 将根本没有 Partitioner。
正如 Raphael 提到的，我也不明白 preserving partitioning 的确切含义。我有这样的代码：data.mapPartitions({iter => iter.flatMap(sth..)}, preservesPartitioning = true); 使用这个 flatMap 我完全可以使用不同的分区器获得不同类型的 RDD 那么preservesPartitioning = true 意味着什么？
通过将preservesPartitioning 设置为true 你告诉Spark 在RDD 上的操作没有改变键（元组的第一个元素）因此Partitioner 的父级（如果存在）对孩子仍然有效。
RDD[(Array[Int], Int)] cannot use (Hash)Partitioner （除非你自己定义），所以无论如何它都会被忽略。但是，如果您更改分区数据上的键，但 preservesPartitioning = true 您会误导 Spark，并且下游处理可能会产生不正确的结果。所以永远不要那样做。

【解决方案2】：

过滤器确实保留了分区，至少过滤器的源代码（preservesPartitioning = true）建议这样做：

  /**
   * Return a new RDD containing only the elements that satisfy a predicate.
   */
  def filter(f: T => Boolean): RDD[T] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[T, T](
      this,
      (context, pid, iter) => iter.filter(cleanF),
      preservesPartitioning = true)
  }

【讨论】：