У нас есть приложение для пакетной обработки Spark 1.6.
Это приложение выполняет запрос Hive, который создает объединение нескольких таблиц, а затем выполняет итерацию, хотя DataFrame, созданный этим запросом Hive.
Следующий псевдокод является хорошим приближением к тому, что делается App -
DataFrame runQueries(){
String query = "select metrics.a, metrics.b, metrics.c, org.name "
+ " from metrics join org "
+ " where metrics.orgId = org.orgId ";
sqlContext sql query;
}
RDD process(Dataframe df){
df.flatMap{ r=>{
// Step 1
// Some Processing on row
}
}
}
public static void main(String[] args){
process(runQueries());
}
Это упрощенная версия того, что делает приложение, но это именно то, что делается. В некоторых случаях это приложение завершается сбоем.
Мы видим "java .lang.OutOfMemoryError: G C overhead limit over Превышен" в журналах Executor в шаге 1 выше, и приложение Spark завершается сбоем, когда это происходит , Ниже приведена соответствующая часть журналов исполнителя.
8315.317: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932129K(4587520K)] [ParOldGen: 5242648K->5242648K(5242880K)] 9174808K->9174777K(9830400K), [Metaspace: 70537K->70537K(1112064K)], 13.4313660 secs] [Times: user=196.04 sys=0.40, real=13.43 secs]
8328.749: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932134K(4587520K)] [ParOldGen: 5242648K->5242648K(5242880K)] 9174808K->9174782K(9830400K), [Metaspace: 70537K->70537K(1112064K)], 14.1754076 secs] [Times: user=188.44 sys=0.61, real=14.17 secs]
8342.925: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932137K(4587520K)] [ParOldGen: 5242648K->5242648K(5242880K)] 9174808K->9174785K(9830400K), [Metaspace: 70537K->70537K(1112064K)], 13.8045601 secs] [Times: user=194.92 sys=0.27, real=13.81 secs]
8356.730: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3931860K(4587520K)] [ParOldGen: 5242648K->5242598K(5242880K)] 9174808K->9174458K(9830400K), [Metaspace: 70537K->70354K(1112064K)], 16.3985571 secs] [Times: user=230.08 sys=0.71, real=16.40 secs]
#
# java.lang.OutOfMemoryError: GC overhead limit exceeded
# -XX:OnOutOfMemoryError="kill %p"
# Executing /bin/sh -c "kill 7055"...
8373.133: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3931939K(4587520K)] [ParOldGen: 5242598K->5242565K(5242880K)] 9174758K->9174504K(9830400K), [Metaspace: 70356K->70356K(1112064K)], 14.7777432 secs] [Times: user=183.87 sys=0.47, real=14.78 secs]
8387.912: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932050K(4587520K)] [ParOldGen: 5242565K->5242564K(5242880K)] 9174725K->9174614K(9830400K), [Metaspace: 70359K->70359K(1112064K)], 15.8329129 secs] [Times: user=233.87 sys=0.55, real=15.83 secs]
8403.745: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932050K(4587520K)] [ParOldGen: 5242564K->5242564K(5242880K)] 9174724K->9174614K(9830400K), [Metaspace: 70359K->70359K(1112064K)], 12.6728274 secs] [Times: user=202.52 sys=0.26, real=12.68 secs]
8416.419: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932063K(4587520K)] [ParOldGen: 5242564K->5242564K(5242880K)] 9174724K->9174627K(9830400K), [Metaspace: 70359K->70359K(1112064K)], 13.3367124 secs] [Times: user=205.25 sys=0.26, real=13.33 secs]
8429.756: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932084K(4587520K)] [ParOldGen: 5242564K->5242564K(5242880K)] 9174724K->9174648K(9830400K), [Metaspace: 70359K->70359K(1112064K)], 12.7089572 secs] [Times: user=203.63 sys=0.25, real=12.71 secs]
8442.465: [Full GC (Ergonomics) [PSYoungGen: 3932160K->3932084K(4587520K)] [ParOldGen: 5242564K->5242564K(5242880K)] 9174724K->9174648K(9830400K), [Metaspace: 70360K->70360K(1112064K)], 13.4739778 secs] [Times: user=205.06 sys=0.35, real=13.48 secs]
8455.940: [Full GC (Ergonomics) [PSYoungGen: 3932160K->0K(4587520K)] [ParOldGen: 5242564K->217955K(5242880K)] 9174724K->217955K(9830400K), [Metaspace: 70360K->70360K(1112064K)], 0.2284579 secs] [Times: user=0.64 sys=0.01, real=0.23 secs]
2020-04-09 05:19:59 ERROR CoarseGrainedExecutorBackend:57 - RECEIVED SIGNAL 15: SIGTERM
Очевидно, что потребление памяти этим исполнителем составляет 99%
Это происходит только для одного исполнителя в прогонах, которые не пройдены. Остальные исполнители имеют общую загрузку ~ 85%. Мы также добавили экспортер метрик в это приложение и видим то же самое в Grafana - при неудачных запусках лишь немногие исполнители нарушают потребление памяти на 90%.
Мы видим эту проблему только для отдельных запусков этого приложения. В частности, мы наблюдаем это для DF с ~ 800M строками.
Ниже приведена конфигурация -
- spark.executor.instances = 500
- spark .executor.memory = 10Gb
- spark. sql .shuffle.partitions = 500
Мы не указываем уровень персистентности для СДР.
Мы новички в Spark и пытаемся угадать проблему. Мы думаем, что это может быть потому, что данные не распределены равномерно по исполнителям, хотя мы не уверены.
Если проблема с перераспределением данных между исполнителями действительно является проблемой, мы рассматриваем следующие исправления -
- Вызывает перераспределение (500 * n) на неисправном СДР. Это приведет к полному перемешиванию, и на больших наборах данных гарантируется равномерное распределение данных по разделам. Кроме того, и, насколько мы понимаем, ожидается, что только один Исполнитель вызовет требуемую операцию на одном Разделе СДР. Таким образом, если мы уменьшаем данные в каждом разделе, мы также считываем объем данных, который каждый исполнитель должен хранить в памяти. Это правильное чтение? Даже если это сработало сейчас, это не гарантирует работу для больших наборов данных. Что если у нас, скажем, 10 миллиардов строк? Выбранное нами значение n не будет работать для этого набора данных. Как мы можем обойти эту проблему?
- Выберите уровень хранения MEMORY_AND_DISK_SER. Насколько мы можем судить, в нем будут храниться блоки RDD, которые не помещаются в память на диск. Это замедлит обработку, но это гораздо лучше, чем полный сбой для нашего варианта использования.
Правильно ли мы считаем, что причиной сбоев является неравномерное распределение данных между исполнителями? Есть ли что-нибудь еще, что можно сделать здесь, кроме этих двух исправлений?
Спасибо, что нашли время прочитать этот длинный пост.