Apache Spark如何缓存Dataset避免转换逻辑被重复执行

你的理解完全正确。这是Spark惰性求值机制的正常表现：所有转换算子（如map、filter）仅记录计算逻辑，不会立刻执行。每次调用行动算子（此处的count()属于行动算子）时，都会从血缘链路的源头开始完整执行一遍计算流程。你连续调用了两次count()，对应两条独立的执行链路，所以cost极高的map逻辑会被执行两次。

解决方案 #

方案1：对中间结果integerDataSet做缓存（最适配当前场景）

你只需要在定义完integerDataSet后新增一行缓存逻辑，即可让map的结果仅计算一次：

// 可选导入存储级别类：import org.apache.spark.storage.StorageLevel;
final Dataset<Integer> integerDataSet = someVeryBigDataSet.map((MapFunction<Row, Integer>) row -> someSuperComplexAndHeavyComputationThatShouldBeDoneOnlyOnceToConvertRowToInteger(row), Encoders.INT());
// 新增缓存逻辑：MEMORY_AND_DISK表示内存不足时自动溢写到磁盘，避免OOM
integerDataSet.persist(StorageLevel.MEMORY_AND_DISK());

第一次调用count()时，map的计算结果会被缓存到内存/磁盘中，第二次调用count()会直接读取缓存结果，不会再次触发map计算。
所有依赖integerDataSet的操作执行完成后，你可以手动调用integerDataSet.unpersist()主动释放缓存资源，也可以等待作业结束后由Spark自动清理。

方案2：使用checkpoint持久化结果（适合超大计算量、跨作业复用场景）

如果你的计算结果需要跨作业复用，或者数据量极大、缓存可靠性不足，可以用checkpoint机制将结果持久化到可靠存储中：

// 先设置checkpoint存储目录，支持本地路径或HDFS路径
sparkSession.sparkContext().setCheckpointDir("/tmp/spark_checkpoint");
final Dataset<Integer> integerDataSet = someVeryBigDataSet.map((MapFunction<Row, Integer>) row -> someSuperComplexAndHeavyComputationThatShouldBeDoneOnlyOnceToConvertRowToInteger(row), Encoders.INT());
// 标记该数据集需要做checkpoint
integerDataSet.checkpoint();

checkpoint会完全切断该数据集的上游血缘，后续所有依赖该数据集的计算都会直接读取持久化的结果，不会重跑上游逻辑。

额外优化建议 #

针对你的统计需求，完全可以通过一次聚合操作完成统计，不需要拆分两次filter和两次count，本身就仅需要触发一次计算，连额外的缓存操作都可以省略：

import static org.apache.spark.sql.functions.*;

Dataset<Row> statResult = integerDataSet.agg(
    count(when(col("value").equalTo(0), 1)).alias("good_count"),
    count(when(col("value").notEqual(0), 1)).alias("bad_count")
);
// 一次行动算子获取两个结果
Row result = statResult.head();
long goodCount = result.getAs("good_count");
long badCount = result.getAs("bad_count");
LOGGER.info("good integer dataset size and bad integer dataset size:\n" + goodCount + " " + badCount);

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