数据帧持久化不会提高 Spark 的性能

Question

我正在编写一个Scala脚本，该脚本使用Spark从表中读取，转换数据并显示结果。我正在使用Spark 2.1.1.2和Scala 2.11.8。我在脚本中使用了两次数据帧实例（df2在下面的代码中）。由于数据帧是在对它们调用操作时计算的，而不是在声明它们时计算的，因此我预测该数据帧将被计算两次。我认为保留此数据帧可以提高性能，认为如果持久化，它将计算一次（持久时），而不是两次。

但是，与不坚持时相比，我坚持时脚本运行的持续时间长 ~10 秒。我不知道这是什么原因。如果有人有想法，将不胜感激。

我的提交命令行如下：

spark-submit --class TestQuery --master yarn --driver-memory 10G --executor-memory 10G --executor-cores 2 --num-executors 4 /home/bcp_data/test/target/TestQuery-1.0-SNAPSHOT.jar

Scala 脚本如下：

 val spark = SparkSession
             .builder()
             .appName("TestQuery")
             .config("spark.sql.warehouse.dir", "file:/tmp/hsperfdata_hdfs/spark-warehouse/")
             .enableHiveSupport()
             .getOrCreate()

 val m = spark.sql("select id, startdate, enddate, status from members")
 val l = spark.sql("select mid, no, status, potential from log")
 val r = spark.sql("select mid, code from records")
val df1 =  m.filter(($"status".isin(1,2).and($"startdate" <= one_year_ago)).and((($"enddate" >= one_year_ago)))
val df2 = df1.select($"id", $"code").join(l, "mid").filter(($"status".equalTo(1)).and($"potential".notEqual(9))).select($"no", $"id", $"code")
df2.persist
val df3 = df2.join(r, df2("id").equalTo(r("mid"))).filter($"code".isin("0001","0010","0015","0003","0012","0014","0032","0033")).groupBy($"code").agg(countDistinct($"no"))

val fa = spark.sql("select mid, acode from actions")
val fc = spark.sql("select dcode, fcode from params.codes")
val df5 = fa.join(fc, fa("acode").startsWith(fc("dcode")), "left_outer").select($"mid", $"fcode")
val df6 = df2.join(df5, df2("id").equalTo(df5("mid"))).groupBy($"code", $"fcode")
println("count1: " + df3.count + " count2: " + df6.count)

Answer 1

在这里使用缓存是正确的选择，但是您的陈述

df2.persist

不起作用，因为您没有使用返回的数据帧。只是做

val df2 = df1.select($"id", $"code")
.join(l, "mid")
.filter(($"status".equalTo(1)).and($"potential".notEqual(9)))
.select($"no", $"id", $"code")
.persist