Почему у pyspark заканчивается память только для карты?

Question

У меня есть простые программы pyspark, что читает 2 текстовых файлов одновременно, преобразовать каждую строку в объект JSON и записать его на паркет файл, как это:

for f in chunk(files, 2):     
    file_rdd = sc.textFile(f) 
    df = (file_rdd 
      .map(decode_to_json).filter(None) 
      .toDF(schema) 
      .coalesce(5) 
      .write 
      .partitionBy("created_year", "created_month") 
      .mode("append") 
      .parquet(file_output)) 

я запускаю работу с пряжей и конфигурации, как это:

conf = (SparkConf() 
     .setAppName(app_name) 
     .set("spark.executor.memory", '6g') 
     .set('spark.executor.instances', '6') 
     .set('spark.executor.cores', '2') 
     .set("parquet.enable.summary-metadata", "false") 
     .set("spark.sql.parquet.compression.codec", 'snappy') 
     ) 

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

Answer 1

Spark имеет много движущихся частей. Он считывает данные из текста в разделы (которые обычно хранятся в памяти), вы декодируете json, который может вызвать проблемы, если ваша линия очень длинная (т. Е. Большой json-объект), вы делаете разделBy, который может иметь слишком много элементов.

Я бы начал с попытки увеличить количество разделов для начала (т. Е. Использовать перераспределение, а не объединять, что уменьшает количество разделов), я бы также попытался написать без partitionBy, и если это не поможет найти самый длинный json и попытаться проанализировать его (т. е. нарисовать строку json по длине и взять наиболее длинный).