Сбой задания SparkSQL при вызове stddev более 1000 столбцов

Question

Я на DataBricks с Spark 2.2.1 и Scala 2.11.Я пытаюсь выполнить запрос SQL, который выглядит следующим образом.

select stddev(col1), stddev(col2), ..., stddev(col1300)
from mydb.mytable

Затем я выполняю код следующим образом.

myRdd = sqlContext.sql(sql)

Однако я вижу следующее исключение.

Job aborted due to stage failure: Task 24 in stage 16.0 failed 4 times, most recent failure: Lost task 24.3 in stage 16.0 (TID 1946, 10.184.163.105, executor 3): org.codehaus.janino.JaninoRuntimeException: failed to compile: org.codehaus.janino.JaninoRuntimeException: Constant pool for class org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificMutableProjection has grown past JVM limit of 0xFFFF
/* 001 */ public java.lang.Object generate(Object[] references) {
/* 002 */   return new SpecificMutableProjection(references);
/* 003 */ }
/* 004 */
/* 005 */ class SpecificMutableProjection extends org.apache.spark.sql.catalyst.expressions.codegen.BaseMutableProjection {
/* 006 */
/* 007 */   private Object[] references;
/* 008 */   private InternalRow mutableRow;
/* 009 */   private boolean evalExprIsNull;
/* 010 */   private boolean evalExprValue;
/* 011 */   private boolean evalExpr1IsNull;
/* 012 */   private boolean evalExpr1Value;
/* 013 */   private boolean evalExpr2IsNull;
/* 014 */   private boolean evalExpr2Value;
/* 015 */   private boolean evalExpr3IsNull;
/* 016 */   private boolean evalExpr3Value;
/* 017 */   private boolean evalExpr4IsNull;
/* 018 */   private boolean evalExpr4Value;
/* 019 */   private boolean evalExpr5IsNull;
/* 020 */   private boolean evalExpr5Value;
/* 021 */   private boolean evalExpr6IsNull;

Отслеживание стека просто продолжается и продолжается, и даже ноутбук Databricks падает из-за многословия.Кто-нибудь когда-нибудь видел это?

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

select avg(col1), ..., avg(col1300) from mydb.mytable
select percentile_approx(col1, 0.5), ..., percentile_approx(col1300, 0.5) from mydb.mytable

Проблема, похоже, связана с stddev но исключение не полезно.Есть идеи о том, что происходит?Есть ли другой способ легко вычислить стандартное отклонение, которое не приведет к этой проблеме?

Оказывается, что post описывает ту же проблему, говоря, что Spark не может обрабатывать широкие схемы илимного столбцов из-за ограничения классов размером 64 КБ.Однако, если это так, то почему avg и percentile_approx работают?

Answer 1

Несколько вариантов:

• Попробуйте отключить генерацию кода всего этапа:

  spark.conf.set("spark.sql.codegen.wholeStage", false)
  

• Если приведенное выше не помогает переключиться наСДР (принят из этого ответа zeo323 ):

  import org.apache.spark.mllib.linalg._
  import org.apache.spark.mllib.stat.MultivariateOnlineSummarizer
  
  val columns: Seq[String] = ???
  
  df
    .select(columns map (col(_).cast("double")): _*)
    .rdd
    .map(row => Vectors.dense(columns.map(row.getAs[Double](_)).toArray))
    .aggregate(new MultivariateOnlineSummarizer)(
       (agg, v) => agg.add(v), 
       (agg1, agg2) => agg1.merge(agg2))
  

• Соберите столбцы в один вектор, используя VectorAssembler и используйтеAggregator, аналогично тому, который использовался здесь , с настройкой метода finish (вам может потребоваться дополнительная настройка для преобразования ml.linalg.Vectors в mllib.linalg.Vectors).

Тем не менее, если это так, то почему работают avg и процентиль_approx?

Spark буквально генерирует Java-код для этих этапов.Поскольку логика не совпадает, размер выходного документа будет отличаться.