RDD/Dataframe 的分区位置

Question

我有一个（相当大，想想 10e7 行）DataFrame，我根据一些属性

从中过滤元素

val res = data.filter(data(FieldNames.myValue) === 2).select(pk.name, FieldName.myValue)

我的DataFrame有n个分区data.rdd.getNumPartitions

现在我想知道我的行来自哪个分区。我知道我可以用这样的东西遍历所有分区

val temp = res.first() //or foreach, this is just an example
data.foreachPartition(f => {
    f.exists(row => row.get(0)==temp.get(0))
    //my code here
}) //compare PKs

或data.rdd.mapPartitionsWithIndex((idx, f) => ...)

但是，如果我的结果和我的 DataFrame 变大，这似乎过分而且性能也不是很好。

在我执行 filter() 操作后，是否有 Spark 方法来执行此操作？

或者，是否有一种方法可以重写/替代 filter() 语句，使其 returns 成为行的来源？

我也可以将分区位置保存在我的 DataFrame 中并在重新分区时更新它，但我宁愿以火花方式进行

（我发现的唯一类似问题是here, and neither the question nor the comment is very helpful. I also found ，可能相似但不相同）

在此先感谢任何help/pointers，如果我错过了一个已经回答过的类似于我的问题，我深表歉意。

Answer 1

分区numbers/counts不稳定，因为Spark会在分区中执行自动扩展和缩减。这意味着输入分区计数可能与输入文件计数不同，例如。

这些情况下的一般模式是根据每个输入文件中的数据创建某种类型的复合键。如果密钥很大，您可以对其进行散列以减小大小。如果您不太关心碰撞，请使用 Murmur3。如果担心冲突，使用MD5，还是挺快的。

如果您拥有的唯一独特特征是输入文件的路径，则必须添加文件路径作为区分列。这是一种方法：

val paths = Seq(...)
val df = paths
  .map { path => 
    sqlContext.read.parquet(path)
      .withColumn("path", lit(path))
  }
  .reduceLeft(_ unionAll _)

想法很简单：一次读取一个输入文件，添加一个与它们关联的唯一列，然后使用 UNION ALL.

将它们组合在一起

RDD/Dataframe 的分区位置

Partition Location of RDD/Dataframe

hadoop-partitioning

apache-spark

rdd

apache-spark-sql

spark-dataframe