избежать перегородки дисбаланс Spark

Question

У меня есть проблема с производительностью с кодом, который я редактирую, каждый раз будет давать OOM при выполнении счета. Я думаю, что я нашел проблему, в основном после keyBy tranformation, исполняемый aggregateByKey. Проблема заключается в том, что почти 98% элементов RDD имеют один и тот же ключ, поэтому aggregationByKey генерирует shuffle, помещает почти все записи в один и тот же перегородка, нижняя строка: всего несколько исполнителей работает, и для этого требуется много давления памяти.

Это код:

val rddAnomaliesByProcess : RDD[AnomalyPO] = rddAnomalies 
    .keyBy(po => po.getProcessCreator.name) 
    .aggregateByKey(List[AnomalyPO]())((list,value) => value +: list,_++_) 
    .map {case(name,list) => 
     val groupByKeys = list.groupBy(po => (po.getPodId, po.getAnomalyCode, po.getAnomalyReason, po.getAnomalyDate, po.getMeasureUUID)) 
     val lastOfGroupByKeys = groupByKeys.map{po => (po._1, List(po._2.sortBy { po => po.getProcessDate.getMillis }.last))} 
     lastOfGroupByKeys.flatMap(f => f._2) 
    } 
    .flatMap(f => f) 

log.info("not duplicated Anomalies: " + rddAnomaliesByProcess.count) 

Я бы способ выполнить операцию в более параллельно, что позволяет всем исполнителям работать почти одинаково. Как я могу это сделать?

Должен ли я использовать пользовательский разделитель?

Answer 1

Если наблюдение верно и

98% элементов RDD имеет тот же ключ

тогда изменение секционирования не поможет вам на всех. По определению разделителя 98% данных должны обрабатываться одним исполнителем.

К счастью, плохой код, вероятно, представляет большую проблему, чем перекос. Пропустив через:

.aggregateByKey(List[AnomalyPO]())((list,value) => value +: list,_++_) 

который просто народная магия, похоже, весь трубопровод можно переписать в виде простого reuceByKey. Псевдокод:

• Объединить name и локальные ключи в одном ключе:

  def key(po: AnomalyPO) = (
      // "major" key 
      po.getProcessCreator.name, 
      // "minor" key 
      po.getPodId, po.getAnomalyCode, 
      po.getAnomalyReason, po.getAnomalyDate, po.getMeasureUUID 
  ) 
  

  ключ, содержащий имя, дата и дополнительные поля должны иметь гораздо более высокую мощность, чем только имя.

• Карта пар и сократить ключ:

  rddAnomalies 
      .map(po => (key(po), po)) 
      .reduceByKey((x, y) => 
      if(x.getProcessDate.getMillis > y.getProcessDate.getMillis) x else y 
  )