apache-spark - Spark 分区 - 使用 DISTRIBUTE BY 选项

Question

我们有一个应该处理 50MM 行的 Spark 环境。这些行包含一个关键列。 key 的唯一数量接近 2000。我想并行处理所有这 2000 个 key 。因此我们使用了如下的 Spark sql

hiveContext.sql("select * from BigTbl DISTRIBUTE by KEY")

随后我们有一个 mapPartitions，它可以很好地并行处理所有分区。然而问题是，它默认只创建 200 个分区。使用如下命令，我可以增加分区

hiveContext.sql("set spark.sql.shuffle.partitions=500");

然而，在实际生产运行期间，我不知道唯一键的数量是多少。我希望这是自动管理的。请问有没有办法做到这一点。

谢谢

巴拉

Answer 1

我建议您使用“重新分区”功能，然后将重新分区的内容注册为新的临时表，并进一步缓存它以加快处理速度。

val distinctValues = hiveContext.sql("select KEY from BigTbl").distinct().count() // find count distinct values 

hiveContext.sql("select * from BigTbl DISTRIBUTE by KEY")
       .repartition(distinctValues.toInt) // repartition to number of distinct values
       .registerTempTable("NewBigTbl") // register the repartitioned table as another temp table

hiveContext.cacheTable("NewBigTbl") // cache the repartitioned table for improving query performance

进一步查询使用“NewBigTbl”