Кэширование таблицы temp с искровым sql

Является ли таблица зарегистрированной в registerTempTable (createOrReplaceTempViewс искру 2. +) кешированной?

Используя Zeppelin, я регистрирую DataFrame в моем scala коде, после тяжелых вычислений, а затем в %pyspark я хочу получить к нему доступ и дополнительно отфильтровать его.

Будет ли он использовать кэшированную версию таблицы? Или он будет перестроен каждый раз?

Ответы

Ответ 1

Зарегистрированные таблицы не кэшируются в памяти.

Метод registerTempTable createOrReplaceTempView просто создаст или заменит представление данного DataFrame заданным планом запроса.

Он преобразует план запроса в канонизированную строку SQL и сохраняет его как текст представления в metastore, если нам нужно создать постоянное представление.

Вам необходимо явно кэшировать ваш DataFrame. например:

df.createOrReplaceTempView("my_table") # df.registerTempTable("my_table") for spark <2.+
spark.cacheTable("my_table")

EDIT:

Давайте проиллюстрируем это на примере:

Использование cacheTable :

scala> val df = Seq(("1",2),("b",3)).toDF
// df: org.apache.spark.sql.DataFrame = [_1: string, _2: int]

scala> sc.getPersistentRDDs
// res0: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> df.createOrReplaceTempView("my_table")

scala> sc.getPersistentRDDs
// res2: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> spark.catalog.cacheTable("my_table") // spark.cacheTable("...") before spark 2.0

scala> sc.getPersistentRDDs
// res4: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map(2 -> In-memory table my_table MapPartitionsRDD[2] at cacheTable at <console>:26)

Теперь тот же пример с использованием cache.registerTempTable cache.createOrReplaceTempView :

scala> sc.getPersistentRDDs
// res2: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> val df = Seq(("1",2),("b",3)).toDF
// df: org.apache.spark.sql.DataFrame = [_1: string, _2: int]

scala> df.createOrReplaceTempView("my_table")

scala> sc.getPersistentRDDs
// res4: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> df.cache.createOrReplaceTempView("my_table")

scala> sc.getPersistentRDDs
// res6: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = 
// Map(2 -> ConvertToUnsafe
// +- LocalTableScan [_1#0,_2#1], [[1,2],[b,3]]
//  MapPartitionsRDD[2] at cache at <console>:28)

Ответ 2

Это не так. Вы должны кэшировать явно:

sqlContext.cacheTable("someTable")