Необходимо ли кэширование для фрейма данных, который повторно используется перед первым действием?

Question

У меня есть фрейм данных, который я по-разному преобразовываю, прежде чем присоединить результат к финальному DF. Промежуточные преобразованные кадры данных никогда не используются ни в каких «действиях». Первое действие когда-либо вызывается только после того, как все части соединены вместе. Мой вопрос - я должен кэшировать первый кадр данных тогда? Пример:

arpu_df=get_arpu_df(..). #.cache() will help here?
sample_by_arpu_ranges=arpu_df.filter("arpu>50").sample(False,0.4)\
    .union(
        arpu_df.filter("arpu>20 and arpu<=50").sample(False,0.1)
        )\
    .union(
        arpu_df.filter("arpu<=20").sample(False,0.02)
        ).select("base_subsc_id")
sample_by_arpu_ranges.count()

sample это трансформация, насколько я знаю. Интересно, будет ли пересчитана часть arpu_df для применения каждого из фильтров, или построитель логического плана поймет, что он может повторно использовать ее в разных частях плана?

Answer 1

Кэш сработает только после вызова действия, поэтому в вашем случае ответ будет нет , cache не будет полезным перед вызовом sample_by_arpu_ranges.count().Обычный обходной путь - вызвать менее дорогостоящее действие, которое является count() сразу после cache(), тогда ваш код будет выглядеть следующим образом:

arpu_df=get_arpu_df(..)

arpu_df.cache()
arpu_df.count()

sample_by_arpu_ranges=arpu_df.filter("arpu>50").sample(False,0.4)\
    .union(
        arpu_df.filter("arpu>20 and arpu<=50").sample(False,0.1)
        )\
    .union(
        arpu_df.filter("arpu<=20").sample(False,0.02)
        ).select("base_subsc_id")
sample_by_arpu_ranges.count()

Answer 2

Ответ находится внутри вашего вопроса.У вас есть только одно действие, поэтому все ваши преобразования будут выполнены в это время.В этом случае вам не нужно сохранять (или кэшировать) ваш фрейм данных.

Persist полезен только в том случае, если вам нужно будет снова вычислить преобразования

пример:

arpu_df=get_arpu_df(..)
sample_by_arpu_ranges=arpu_df.filter("arpu>50").sample(False,0.4)\
    .union(
        arpu_df.filter("arpu>20 and arpu<=50").sample(False,0.1)
        )\
    .union(
        arpu_df.filter("arpu<=20").sample(False,0.02)
        ).select("base_subsc_id").persist() //here you persist sample_by_arpu_ranges because you know you will have multiple actions on it

sample_by_arpu_ranges.count() // 1st action

sample_by_arpu_ranges.write.parquet("path") // 2nd action  

В этом примере sample_by_arpu_ranges будет сохраняться во время 1-го действия, таким образом, длявторое действие, sample_by_arpu_ranges будет готово.

-> Без сохранения одним действием:

arpu_df = spark.read.parquet(path) 
sample_by_arpu_ranges=arpu_df.filter(...)
sample_by_arpu_ranges.count()

что происходит:

• sample_by_arpu_ranges.count ()
• arpu_df = spark.read.parquet (путь)
• sample_by_arpu_ranges = arpu_df.filter (...)
• count

-> не сохраняет arpu_df или sample_by_arpu_ranges, но вы не делаетебольше не нужно

-> Без сохранения нескольких действий:

arpu_df = spark.read.parquet(path) 
sample_by_arpu_ranges=arpu_df.filter(...)
arpu_df.count()
sample_by_arpu_ranges.count()

что происходит:

• arpu_df.count ()
• arpu_df = spark.read.parquet (путь)
• count

-> не сохраняет arpu_df!

• sample_by_arpu_ranges.count ()

• arpu_df = spark.read.parquet (путь) // Вы должны прочитать снова!

• sample_by_arpu_ranges = arpu_df.filter(...)

• count

-> С сохранением с несколькими действиями:

arpu_df = spark.read.parquet(path).persist
sample_by_arpu_ranges=arpu_df.filter(...)
arpu_df.count()
sample_by_arpu_ranges.count()

что происходит:

• arpu_df.count ()
• arpu_df = spark.read.parquet (path)
• persist ---> сохранить arpu_df в кеш
• count
• sample_by_arpu_ranges.count ()
• sample_by_arpu_ranges= arpu_df (взять из кеша, читать не нужно) .filter (...)
• count