Я представляю собой приложение Spark, в котором есть несколько точек, в которых я хотел бы сохранить текущее состояние. Обычно это происходит после большого шага или кэширования состояния, которое я хотел бы использовать несколько раз. Похоже, что когда я вызываю кэш в своем фрейме данных во второй раз, новая копия кэшируется в памяти. В моем приложении это приводит к проблемам с памятью при масштабировании. Несмотря на то, что в моих текущих тестах размер данного кадра данных составляет максимум около 100 МБ, совокупный размер промежуточных результатов выходит за пределы выделенной памяти исполнителя. Ниже приведен небольшой пример, демонстрирующий такое поведение.
cache_test.py:
Код: Выделить всё
from pyspark import SparkContext, HiveContext
spark_context = SparkContext(appName='cache_test')
hive_context = HiveContext(spark_context)
df = (
hive_context
.read
.format('com.databricks.spark.csv')
.load('simple_data.csv')
)
df.cache()
df.show()
df = df.withColumn('C1+C2', df['C1'] + df['C2'])
df.cache()
df.show()
spark_context.stop()
simple_data.csv:
В пользовательском интерфейсе приложения помимо копии с новым столбцом есть копия исходного фрейма данных. Я могу удалить исходную копию, вызвав df.unpersist() перед строкой withColumn. Рекомендуемый ли это способ удаления кешированного промежуточного результата (т. е. вызывать unpersist перед каждым кэшем()).
Кроме того, можно ли очистить все кешированные объекты. В моем приложении есть естественные точки останова, где я могу просто очистить всю память и перейти к следующему файлу. Я хотел бы сделать это без создания нового приложения Spark для каждого входного файла.
Заранее спасибо!
Подробнее здесь:
https://stackoverflow.com/questions/369 ... in-pyspark
Мобильная версия