Блог

Я новичок в spark. Я кодирую алгоритм машинного обучения в Spark standalone (v3.0.0) с этим набором конфигураций:

 SparkConf conf = new SparkConf();
conf.setMaster("local[*]");
conf.set("spark.driver.memory", "8g");
conf.set("spark.driver.maxResultSize", "8g");
conf.set("spark.memory.fraction", "0.6");
conf.set("spark.memory.storageFraction", "0.5");
conf.set("spark.sql.shuffle.partitions", "5");
conf.set("spark.memory.offHeap.enabled", "false");
conf.set("spark.reducer.maxSizeInFlight", "96m");
conf.set("spark.shuffle.file.buffer", "256k");
conf.set("spark.sql.debug.maxToStringFields", "100");
  

Вот как я создаю перекрестный валидатор

 ParamMap[] paramGrid = new ParamGridBuilder()
            .addGrid(gbt.maxBins(), new int[]{50})
            .addGrid(gbt.maxDepth(), new int[]{2, 5, 10})
            .addGrid(gbt.maxIter(), new int[]{5, 20, 40})
            .addGrid(gbt.minInfoGain(), new double[]{0.0d, .1d, .5d})
            .build();

    CrossValidator gbcv = new CrossValidator()
            .setEstimator(gbt)
            .setEstimatorParamMaps(paramGrid)
            .setEvaluator(gbevaluator)
            .setNumFolds(5)
            .setParallelism(8)
            .setSeed(session.getArguments().getTrainingRandom());
  

Проблема в том, что когда (в paramGrid) maxDepth равен всего {2, 5}, а MaxIter {5, 20}, все работает просто отлично, но когда это похоже на приведенный выше код, он продолжает регистрироваться:
WARN DAGScheduler: broadcasting large task binary with size xx ,
при xx увеличивается с 1000 КБ до 2,9 Мбайт, что часто приводит к исключению тайм-аута
Какие параметры spark я должен изменить, чтобы избежать этого?

Из-за проблемы с таймаутом рассмотрите возможность изменения следующей конфигурации:

spark.sql.autoBroadcastJoinThreshold до -1.

Это снимет ограничение на размер широковещательной рассылки, которое составляет 10 МБ.

Решение, которое сработало для меня, было:

уменьшение размера задачи => уменьшение обработки данных

Сначала проверьте количество разделов в dataframe через df.rdd.getNumPartitions() После увеличьте разделы: df.repartition(100)

Я получил то же WARN org.apache.spark.scheduler.DAGScheduler: Broadcasting large task binary with size 5.2 MiB , что сработало для меня, я увеличиваю конфигурацию компьютера с 2vCPU, 7,5 ГБ ОЗУ, до 4vCPU 15GBRAM (был создан какой-то файл parquet, но задание так и не было завершено, поэтому я увеличиваю до 8vCPU 32 ГБ ОЗУ (теперь все работает). Это в GCP Dataproc