Искра persist () (тогда действие) действительно сохраняется?

Question

Я всегда понимал, что persist() и cache(), затем действие по активации DAG, рассчитают и сохранят результат в памяти для последующего использования. Многие потоки здесь расскажут вам о необходимости кэширования для повышения производительности часто используемых фреймов данных.

Недавно я сделал тест и был сбит с толку, потому что, похоже, это не так.

    temp_tab_name = "mytablename";
    x = spark.sql("select * from " +temp_tab_name +" limit 10");
    x = x.persist()
    x.count() #action to activate all the above steps
    x.show() #x should have been persisted in memory here, DAG evaluated, no going back to "select..." whenever referred to
    x.is_cached #True
    spark.sql("drop table "+ temp_tab_name);
    x.is_cached #Still true!!
    x.show() # Error, table not found here

Так что мне кажется, что х никогда не вычисляется и сохраняется. Следующая ссылка на x все еще восходит к оценке ее определения DAG "select...". Что-то, что я здесь пропустил?

Answer 1

cache и persist не полностью отделяют результат вычисления от источника.Это просто делает все возможное, чтобы избежать пересчета.Так что, вообще говоря, удаление источника до того, как вы закончите с набором данных, является плохой идеей.

Что может пойти не так в вашем конкретном случае (из головы):

1) show не нужны все записи таблицы, поэтому, возможно, он запускает вычисления только для нескольких разделов.Таким образом, большинство разделов еще не рассчитаны на данный момент.

2) spark нуждается в некоторой вспомогательной информации из таблицы (например, для разбиения)

Answer 2

Ниже приведен правильный синтаксис ... вот дополнительная документация для таблиц "uncaching" => https://spark.apache.org/docs/latest/sql-performance-tuning.html ... и вы можете подтвердить приведенные ниже примеры в пользовательском интерфейсе Spark на вкладке "storage", чтобы увидеть объекты "кэшируются" и "не кэшируются"

# df method
df = spark.range(10)
df.cache() # cache
# df.persist() # acts same as cache
df.count() # action to materialize df object in ram
# df.foreach(lambda x: x) # another action to materialize df object in ram
df.unpersist() # remove df object from ram

# temp table method
df.createOrReplaceTempView("df_sql")
spark.catalog.cacheTable("df_sql") # cache
spark.sql("select * from df_sql").count() # action to materialize temp table in ram
spark.catalog.uncacheTable("df_sql") # remove temp table from ram