Scala Spark RDD、DataSet、Pair RDD 和分区

Question

在 Scala Spark 中，有几种方法可以导致数据分区/重新分区。其中包括partitionBy, coalesce, repartition, and textFile 以及具有分区计数作为参数的其他函数。下面，我使用textFile，其规格至少为 8 个分区。我不希望转换来撤消这些分区。对于要保留的分区，您需要persist 分区结果。但是，map 和 flatMap 等函数不会保留分区。我相信这可能会影响性能。 PairRDDS 具有维护分区的mapValues and flatMapValues。

DataSets and RDDs 是否有与map and flatMap 相同的功能不会破坏分区？

如果我把这一切都搞混了，RDD 和 DataSet 如何维护分区，记住 map 和 flatMap 操作是它们操作的关键。

val tweets:RDD[Tweet] = mySpark.sparkContext.textFile(path,8).map(parseTweet).persist()
val numerical_fields_Tweets:Dataset[Tweet] = tweets.toDS()

Answer 1

在 Spark 中，不对数据进行重新分区/洗牌的操作会保留分区（通过对先前建立的分区进行操作）。 map 和 flatMap 就是这样的操作：它们不会改变分区的数量。此外，map 不会更改分区内的行数或其顺序。

RDD 和 DataSet 如何维护分区

您混合了两个概念：(1) 与数据在其转换中的某个点关联的分区器，以及 (2) 数据被拆分成的分区。

数据的分区方式与与数据关联的分区器之间存在差异。如上所述，map 和 flatMap 不会更改分区的数量，但它们不保证与数据关联的分区器。考虑 RDD 的map：

/**
 * Return a new RDD by applying a function to all elements of this RDD.
 */
def map[U: ClassTag](f: T => U): RDD[U] = withScope {
  val cleanF = sc.clean(f)
  new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
}

和MapPartitionsRDD:

private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag](
    var prev: RDD[T],
    f: (TaskContext, Int, Iterator[T]) => Iterator[U],
    preservesPartitioning: Boolean = false, ...)

因此，虽然map 不会重新分区数据，但它不保证与数据关联的分区器，因为map 可以修改行的方式没有限制。

Pair RDD，即RDD[(K, V)]，有些特殊，因为它们通常是分区操作的结果，如果我们使用mapValues而不是map，我们可以确定分区器没有改变因为我们还没有碰过“钥匙”。

/**
 * Pass each value in the key-value pair RDD through a map function without changing the keys;
 * this also retains the original RDD's partitioning.
 */
def mapValues[U](f: V => U): RDD[(K, U)] = self.withScope {
  val cleanF = self.context.clean(f)
  new MapPartitionsRDD[(K, U), (K, V)](self,
    (context, pid, iter) => iter.map { case (k, v) => (k, cleanF(v)) },
    preservesPartitioning = true)
}

希望这会有所帮助！