Невозможно сохранить большой фрейм данных искры в качестве маринада

Question

У меня большой фрейм данных (чуть больше 20G), я пытаюсь сохранить его как объект pickle для последующего использования в другом процессе.

Я пробовал другую конфигурацию, ниже приведены последние.

executor_cores=4                                                                
executor_memory='20g'                                                           
driver_memory='40g'                                                                
deploy_mode='client'                                                            
max_executors_dynamic='spark.dynamicAllocation.maxExecutors=400'                   
num_executors_static=300                                                        
spark_driver_memoryOverhead='5g'                                                 
spark_executor_memoryOverhead='2g'                                               
spark_driver_maxResultSize='8g'                                                 
spark_kryoserializer_buffer_max='1g'                                            

Примечание: - Я не могу увеличить размер spark_driver_maxResultSize больше, чем на 8G.

Я также попытался сохранить фрейм данных в виде файлов hdfs, а затем попытался сохранить его как pickel, но получил сообщение об ошибке, аналогичное предыдущему.

Насколько я понимаю, когда мы используем pandas.pickle, он объединяет все данные в один драйвер, а затем создает объект pickle.Поскольку размер данных больше, чем driver_max_result_size, код не работает.(Код работал ранее для данных 2G).

Есть ли у нас какое-либо решение для решения этой проблемы?

big_data_frame.toPandas().to_pickle('{}/result_file_01.pickle'.format(result_dir))

big_data_frame.write.save('{}/result_file_01.pickle'.format(result_dir), format='parquet', mode='append')

df_to_pickel=sqlContext.read.format('parquet').load(file_path) 
df_to_pickel.toPandas().to_pickle('{}/scoring__{}.pickle'.format(afs_dir, rd.strftime('%Y%m%d')))

Сообщение об ошибке

Py4JJavaError: Произошла ошибка при вызовеo1638.collectToPython.: org.apache.spark.SparkException: задание прервано из-за сбоя этапа: общий размер сериализованных результатов 955 задач (4,0 ГБ) больше, чем spark.driver.maxResultSize (4,0 ГБ)

Answer 1

Сохранение в виде файла выбора - это функция RDD в Spark, а не датафрейм. Чтобы сохранить кадр с помощью pickle, запустите

big_data_frame.rdd.saveAsPickleFile(filename)

Если вы работаете с большими данными, никогда не стоит запускать сбор или toPandas в искре, так как он собирает все в памяти, что приводит к сбою системы. Я бы посоветовал вам использовать паркет или любой другой формат для сохранения ваших данных, поскольку функции СДР находятся в режиме обслуживания, что означает, что спарк не вводит в него какие-либо новые функции.

Чтобы прочитать файл, попробуйте

pickle_rdd = sc.pickleFile(filename).collect()
df = spark.createDataFrame(pickle_rdd)