Spark 分区 - 使用 DISTRIBUTE BY 选项

Question

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

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

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

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

然而在实际生产中运行我不知道唯一键的数量是多少。我希望这是自动管理的。请问有什么办法吗

谢谢

巴拉

Answer 1

我建议您使用 "repartition" 函数，然后将重新分区的文件注册为新临时文件 table 并进一步缓存它以加快处理速度。

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"

Spark 分区 - 使用 DISTRIBUTE BY 选项

Spark partitions - using DISTRIBUTE BY option

hive

hadoop-partitioning

apache-spark