Spark 2.3 Dataframe partition-想要在 n 个分区中对 key 上的数据进行分区答案

【问题标题】：Spark 2.3 Dataframe partition-want to partition data on key in n number of partitionSpark 2.3 Dataframe partition-想要在 n 个分区中对 key 上的数据进行分区
【发布时间】：2020-03-13 08:20:05
【问题描述】：

我需要 spark(scala) 数据帧分区方面的帮助。我需要按一个键列将分区划分为 n 个分区，与同一键相关的所有行都应该在同一个分区中（即键不应该跨分区分布）

注意：我的密钥可能有数百万个

例如：假设我有以下数据框

等等

如您所见，许多值共享相同的键。我想将此数据集划分为“n”个分区，其中相同的键应该在同一个分区中，并且键不应该跨分区分布。多个键位于同一分区并且键不可排序。

提前致谢

【问题讨论】：

使用 partitionBy(colName) 保存dataFrame

标签： scala apache-spark apache-spark-sql partitioning

【解决方案1】：

根据Spark - The Definitive Guide这本书，Spark 有两个内置的分区器，HashPartitioner 用于离散值，RangePartitioner。两者都适用于离散值和连续值。

HashPartitioner 示例：

import org.apache.spark.HashPartitioner

val rdd = df.rdd // convert DataFrame to low-level RDD
val keyedRDD = rdd.keyBy(...) // define your custom key
keyedRDD.partitionBy(new HashPartitioner(n))

分区器示例：

import org.apache.spark.Partitioner

class DomainParitioner extends Partitioner {
  def numPartitions = n
  def getPartition(key: Any): Int = {
    // your custome partition logic
  }
}

keyedRDD.partitionBy(new DomainPartitioner).map(_._1).glom().map(_.toSet.toSeq.length)

书中还提到你应该注意key skew，这意味着一些键可能比其他键有很多很多的值。您希望尽可能多地破坏这些密钥以提高并行性并防止在执行过程中出现OutOfMemoryErrors。

【讨论】：

谢谢 Mike，我猜 hashpartitioner 会起作用，但我的数据框非常大而且很复杂，有什么方法可以在数据框目录上实现这一点，因为在 rdd 中转换然后返回到数据框会耗费时间和资源.
很遗憾，自定义分区器在结构化 API 中不可用。
我实际上是在阅读博客 (acadgild.com/blog/partitioning-in-spark)，这里他们给出了一个自定义分区的示例，但必须为我的问题寻找实现
这个博客看起来很不错，似乎很有帮助。简单看一下，似乎这一切都基于 RDD，并反映了我回答中的示例。

【解决方案2】：

您可以在将 DataFrame 写入基于文件的输出时对其进行分区。比如：

df.write.partitionBy("colName").format("parquet").save(path-to-file)

这将创建您的目录结构

path
└── to
    └── file
         ├── colName=value1
                       └── data.parquet
         ├── colName=value2
                       └── data.parquet

当您加载数据和过滤器时，谓词将被下推到源文件中，您可以获得分区的性能优势

这不是你要找的吗？

【讨论】：

【解决方案3】：

试试

def repartition(partitionExprs: org.apache.spark.sql.Column*): org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]


val df = Seq(("aa","vv"),("aa","v1v1"),("a1","v2")).toDF("Key","Value")
 val partionedDf = df.repartition(col("Key"))

【讨论】：