Ошибка искры из-за обновления файлов S3. Как устранить эту ошибку?

Question

Мой сценарий Spark терпит неудачу, потому что корзина S3, из которой извлекается df, обновляется новыми файлами во время выполнения сценария.Меня не волнуют вновь поступающие файлы, но, видимо, это делает Spark.

Я пытался добавить команду REFRESH TABLE в сообщении об ошибке, но это не работает, потому что это невозможно узнать при исполнениивремя поступления новых файлов, поэтому невозможно узнать, куда поместить эту команду.Я попытался поместить эту команду REFRESH в 4 разных местах в сценарии (другими словами, вызвать ее 4 раза в разных точках в сценарии) - все с тем же сообщением об ошибке

Caused by: java.io.FileNotFoundException: No such file or directory '<snipped for posting>.snappy.parquet'
It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved.

Я создаю dfс: df = spark.table('data_base.bal_daily_posts')

Итак, что я могу сделать, чтобы убедиться, что файлы S3, поступающие на S3 post-script-kickoff, игнорируются и не выдают ошибку в сценарии?

Answer 1

Переместите файлы, которые вы собираетесь обработать, в другую папку (ключ) и наведите курсор на работу только с этой папкой

Answer 2

Я не уверен, что это будет работать или нет, но попробуйте:

Как только вы прочитаете свои входные файлы из S3, немедленно выполните persist операцию с этим фреймом данных, как показано ниже

import org.apache.spark.storage.StorageLevel

val inputDataFrame = sparkSession.read.json("s3a://bucket_name/file_path/)
    .persist(StorageLevel.MEMORY_AND_DISK)

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