在 Spark 中高效地连接一个大表（1TB）和另一个小表（250GB）

Posted 2023-04-15

【中文标题】在 Spark 中高效地连接一个大表（1TB）和另一个小表（250GB）

【英文标题】：Efficiently join one large table(1TB) and other small table(250GB) in spark

【发布时间】：2021-12-30 20:00:39

【问题描述】：

有一个大小为 1TB 的巨大数据集和其他大小为 250GB 的数据集。目前的设置如下

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1) //As I was getting timeoutException

var masterDfPrimary: DataFrame = spark.read //Huge data set
      .format("com.databricks.spark.sqldw")
      .options(readAdwMap) //Connection and table details passed as parameter
      .load().persist(StorageLevel.DISK_ONLY) //Persisted to disk as it will be used twice for computation

masterDfPrimary.createOrReplaceTempView("HugeData")

var smallDF= spark.read.parquet(path)
    smallDF.createOrReplaceTempView("smallData")

var computeDf = spark.sql("""select * from HugeData as h left outer join smallData as s on h.id = s.id)

display(computeDf)

上面的代码是如何对海量数据进行分区的？如何通过使用重新分区和广播概念进一步提高性能，或者我可以利用任何其他配置设置

【问题讨论】：

也许您可以尝试通过碰撞spark.sql.shuffle.partitions 来增加连接的并行度。仍然不清楚您为什么选择保留更大的数据集...

正如我在旁边的 cmets 中已经提到的，稍后我将使用那个更大的数据集两次。如果我不坚持，它会在计算时重新加载数据

【参考方案1】：

您需要使用排序合并连接 由于第二个表大小为 250GB，因此无法广播它 另一种方法是，如果一个小表只有一个实例“id”，那么您可以进行预处理以限制其上表中的记录并将其广播到更大的表以获得更好的性能。