Spark分布式Join优化：如何避免单Executor执行并强制物化数据集？

问题场景与代码 #

现有如下Spark SQL作业代码：

Dataset<Row> currentDataSet = hdfsHandler.loadDataSet(sparkSession, "somelocation");
currentDataSet.createOrReplaceTempView(CURRENT_USER_VIEW);

Dataset<Row> previousDataSet = hdfsHandler.loadDataSet(sparkSession, "somelocation2");
previousDataSet.createOrReplaceTempView(PREVIOUS_USER_VIEW);

String currentRunColumn = "c.".concat("userid");
String previousRunColumn = "p.".concat("userid");

Dataset<Row> addedRecordDataSets = sparkSession.sql("SELECT " + currentRunColumn + " FROM " +
        CURRENT_USER_VIEW + " AS c " +
        " LEFT JOIN " + PREVIOUS_USER_VIEW + " AS p " +
        " ON " + currentRunColumn + " == " + previousRunColumn +
        " WHERE " + previousRunColumn + " IS NULL ");

// 注：原代码中`dataSet`应为`addedRecordDataSets`，以下为修正后逻辑
addedRecordDataSets.coalesce(1).persist(StorageLevel.DISK_ONLY()).foreachPartition(persist());

当前问题 #

该作业会生成3个Spark Job：

• 从HDFS读取currentDataSet
• 从HDFS读取previousDataSet
• 关联两个数据集、执行coalesce(1)并调用persist()

由于foreachPartition(persist())是唯一的终端操作，第三步的所有计算（关联+coalesce）会被推到单Executor节点执行，数据集量大时耗时极久。期望的执行顺序是先分布式完成Join计算，再执行coalesce(1)。

目前想到的方案是在计算addedRecordDataSets后添加count()这类终端操作，但需要缓存数据避免重复执行Join，缓存的存储成本较高。想请教：是否可以仅强制物化addedRecordDataSets，无需额外缓存？

解决方案 #

可以通过以下几种方式实现强制物化数据集，避免单节点瓶颈：

1. 临时缓存+轻量终端操作（低成本物化）

用cache()触发分布式计算，再通过count()强制执行，完成物化后立即释放缓存：

// 触发分布式Join并物化数据集
addedRecordDataSets.cache().count();
// 取消缓存，释放存储资源
addedRecordDataSets.unpersist(false);

// 后续基于物化结果执行coalesce和输出
addedRecordDataSets.coalesce(1).foreachPartition(persist());

这种方式下Join会在分布式节点完成，物化后的数据不会长期占用存储，平衡了性能与成本。

2. Checkpoint持久化到外部存储

如果数据集极大，内存/磁盘缓存压力大，可使用checkpoint()将物化结果写入外部存储（如HDFS），切断依赖链避免重复计算：

// 设置checkpoint目录（需提前在HDFS创建）
sparkSession.sparkContext().setCheckpointDir("/tmp/spark-checkpoint");

// 强制物化数据集，写入外部存储
addedRecordDataSets.checkpoint();

// 后续操作直接读取物化后的结果
addedRecordDataSets.coalesce(1).foreachPartition(persist());

3. 空分布式操作触发物化

通过无逻辑的foreachPartition触发分布式Join计算，无需缓存或外部存储：

// 空操作触发分布式执行，完成数据集物化
addedRecordDataSets.foreachPartition(iter -> {});

// 基于物化结果执行coalesce和输出
addedRecordDataSets.coalesce(1).foreachPartition(persist());

这种方式利用Spark的血统机制，空操作会驱动分布式计算完成，后续的coalesce直接基于物化后的结果执行。

内容的提问来源于stack exchange，提问作者best wishes