Как избежать того, чтобы Spark executor потерялся и контейнер пряжи убил его из-за ограничения памяти?

У меня есть следующий код, который срабатывает hiveContext.sql() большую часть времени. Моя задача состоит в том, чтобы создать несколько таблиц и вставить значения в после обработки для всех разделов таблицы hive.

Поэтому я сначала запускаю show partitions и, используя его выход в цикле for, вызываю несколько методов, которые создают таблицу (если она не существует) и вставляют в них с помощью hiveContext.sql.

Теперь мы не можем выполнить hiveContext в исполнителе, поэтому я должен выполнить это в цикле for В программе драйвера и должен запустить последовательно один на один. Когда я представляю эту работу искры в кластере пряжи, почти все время мой исполнитель теряется из-за перетасовки не найденного исключения.

Теперь это происходит потому, что пряжа убивает моего исполнителя из-за перегрузки памяти. Я не понимаю, почему, поскольку у меня очень маленький набор данных для каждого раздела улья, но все равно это заставляет YARN убивать моего исполнителя.

Будет ли следующий код делать все параллельно и пытаться разместить все данные раздела Улья в памяти одновременно время?

public static void main(String[] args) throws IOException {   

    SparkConf conf = new SparkConf(); 

    SparkContext sc = new SparkContext(conf); 

    HiveContext hc = new HiveContext(sc); 



    DataFrame partitionFrame = hiveContext.sql(" show partitions dbdata partition(date="2015-08-05")"); 

  

    Row[] rowArr = partitionFrame.collect(); 

    for(Row row : rowArr) { 

        String[] splitArr = row.getString(0).split("/"); 

        String server = splitArr[0].split("=")[1]; 

        String date =  splitArr[1].split("=")[1]; 

        String csvPath = "hdfs:///user/db/ext/"+server+".csv"; 

        if(fs.exists(new Path(csvPath))) { 

            hiveContext.sql("ADD FILE " + csvPath); 

        } 

        createInsertIntoTableABC(hc,entity, date); 

        createInsertIntoTableDEF(hc,entity, date); 

        createInsertIntoTableGHI(hc,entity,date); 

        createInsertIntoTableJKL(hc,entity, date); 

        createInsertIntoTableMNO(hc,entity,date); 

   } 

}