DataFrame persist 不会提高 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