Apache Spark: Как сохранить результаты в фрейме данных (фрейм данных с объединениями), чтобы действие на фрейме данных не изменило результаты?

Question

Я кэшировал результаты фрейма данных (dfUpdates); но после MERGE в таблице-A, когда я хочу использовать ранее сохраненные результаты в dfUpdates, результаты отличаются.

Ниже приведено пошаговое объяснение того, что происходит ..

Step-1:
df_Stage.count // 2113983
val df_Prodtable = spark.table("ProdTable") // Delta Table.
df_Prodtable.count  //2112563
val dfUpdates = df_stage.join(df_Prodtable)
                    .filter( Multiple conditions here...) 
                    .cache()
println(dfUpdates.count) //This action should cache the dfUpdates results into Memory. count is 130454, i.e., these many records need to be updated into the Prodtable.

Step-2:
MERGE statement here updates the records into the Prodtable (Delta table) using records from the dfUpdates dataframe. All the 130454 records are getting updated into the Prodtable.

Step-3:
println(dfUpdates.count) // gets zero records.

Я понимаю, что шаг 3 вызывает действие на фрейме данных dfUpdates, и он перевычисляется (выполняется условие соединения пересчитывается). Как сохранить материализованные данные в памяти? так что я могу использовать dfUpdates для последующей обработки.

Answer 1

В вашем случае dfUpdates зависит от ProdTable, поэтому, когда вы обновляете ProdTable во время процесса слияния, spark делает недействительным зависимый кэш dfUpdates, делая счет неправильным.

Вот JIRA , которая может помочь.

При аннулировании кэша мы отключаем другие кэши, зависящие от этого кэша, чтобы обеспечить актуальность кэшированных данных. Например, когда базовая таблица была изменена или сама таблица была удалена, все кэши, которые используют эту таблицу, должны быть признаны недействительными или обновлены.

Чтобы обойти эту ситуацию, следует сохранить промежуточный результат в диск и читай обратно. Это обычная стратегия для разрушения DAG и сохранения повторных вычислений.