apache-spark - 缓存不阻止多次文件扫描？

Question

我对 DataFram APIs cache 的使用有疑问。考虑以下查询:

val dfA = spark.table(tablename)
.cache

val dfC = dfA
.join(dfA.groupBy($"day").count,Seq("day"),"left")

所以 dfA 在这个查询中被使用了两次，所以我认为缓存它会很有用。但是我对这个计划感到困惑，该表仍然被扫描了两次(FileScan 出现了两次):

dfC.explain

== Physical Plan ==
*Project [day#8232, i#8233, count#8251L]
+- SortMergeJoin [day#8232], [day#8255], LeftOuter
   :- *Sort [day#8232 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(day#8232, 200)
   :     +- InMemoryTableScan [day#8232, i#8233]
   :           +- InMemoryRelation [day#8232, i#8233], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas)
   :                 +- *FileScan parquet mytable[day#8232,i#8233] Batched: true, Format: Parquet, Location: InMemoryFileIndex[hdfs://tablelocation], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<day:int,i:int>
   +- *Sort [day#8255 ASC NULLS FIRST], false, 0
      +- *HashAggregate(keys=[day#8255], functions=[count(1)])
         +- Exchange hashpartitioning(day#8255, 200)
            +- *HashAggregate(keys=[day#8255], functions=[partial_count(1)])
               +- InMemoryTableScan [day#8255]
                     +- InMemoryRelation [day#8255, i#8256], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas)
                           +- *FileScan parquet mytable[day#8232,i#8233] Batched: true, Format: Parquet, Location: InMemoryFileIndex[hdfs://tablelocation], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<day:int,i:int>

为什么表没有被缓存？我正在使用 Spark 2.1.1

Answer 1

尝试在缓存之后使用 count()，这样您就可以触发一个操作，并且在“计算”第二个操作的计划之前完成缓存。

据我所知，第一个 Action 会触发缓存，但由于Spark计划不是动态的，如果你缓存后的第一个 Action 使用表两次，它将不得不读取它两次(因为它不会缓存表，直到它执行该操作)。

如果上面的方法不起作用[和/或你遇到了提到的错误]，它可能与计划有关，你也可以尝试将 DF 转换为 RDD，然后再转换回 RDD(这样计划将是100% 准确)。