apache-spark - 避免在没有缓存的情况下对 Spark 中的代码进行延迟评估

Question

如何避免火花中的懒惰评估。我有一个需要立即填充的数据框，因为我需要根据为每一行数据框生成的随机数过滤数据，比如如果生成的随机数 > 0.5，它将被过滤为 dataA，如果生成的随机数 < 0.5 将被过滤为 dataB。

val randomNumberDF = df.withColumn("num", Math.random())
val dataA = randomNumberDF.filter(col("num") >= 0.5)
val dataB = randomNumberDF.filter(col("num") < 0.5)

由于 spark 正在执行惰性评估，因此在过滤时没有可靠的行分布，这些行被过滤为 dataA 和 dataB（有时同一行同时出现在 dataA 和 dataB 中）

如何避免重新计算“num”列，我尝试使用“缓存”，它有效，但考虑到我的数据量会很大，我排除了该解决方案。我还尝试在 randomNumberDF 上使用其他操作，例如：

count
rdd.count
show
first

这些并没有解决问题。

请提出与缓存/持久化/将数据写入 HDFS 不同的建议，然后再次将其作为解决方案读取。

我已经检查过的参考资料：

• 如何强制火花以避免数据集重新计算？
• 如何强制 Spark 只执行一次转换？
• 如何强制 Spark 内联评估 DataFrame 操作

标签： apache-sparkapache-spark-sql

Answer 1

如果您正在寻找一种方法来确保相同的值在 中randomNumberDF.num，那么您可以生成带有种子的随机数（使用org.apache.spark.sql.functions.rand()）：

下面使用 112 作为种子值：

val randomNumberDF = df.withColumn("num", rand(112))
val dataA = randomNumberDF.filter(col("num") >= 0.5)
val dataB = randomNumberDF.filter(col("num") < 0.5)

这将确保 中的值在num的多次评估中是相同的randomNumberDF。