Spark, Как можно добавить больше памяти для хранения?

Hy,

У меня много раз возникает эта ошибка, когда я использую самый большой набор данных и использую MlLib (ALS)

Набор данных имеет 3 столбца (пользователь, фильм и рейтинг) и 1 200 000 строк.

WARN TaskSetManager: Stage 0 contains a task of very large size (116722 KB). The maximum recommended task size is 100 KB.
Exception in thread "dispatcher-event-loop-3" java.lang.OutOfMemoryError: Java heap space

Моя машина теперь имеет 8 ядер, 240 ГБ ОЗУ и 100 ГБ на диске (50 ГБ бесплатно).

Я хочу добавить больше памяти и больше исполнителей, и я установил (я использую spyder IDE)

conf = SparkConf()
conf.set("spark.executor.memory", "40g")
conf.set("spark.driver.memory","20g")
conf.set("spark.executor.cores","8")
conf.set("spark.num.executors","16")
conf.set("spark.python.worker.memory","40g")
conf.set("spark.driver.maxResultSize","0")
sc = SparkContext(conf=conf)

Но у меня все еще есть это:

введите здесь описание изображения

Что я сделал не так?

Как я запускаю Spark (PySpark — Spyder IDE)

import sys
import os
from py4j.java_gateway import JavaGateway
gateway = JavaGateway()


os.environ['SPARK_HOME']="C:/Apache/spark-1.6.0"
sys.path.append("C:/Apache/spark-1.6.0/python/")

from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext

conf = SparkConf()
conf.set("spark.executor.memory", "25g")
conf.set("spark.driver.memory","10g")
conf.set("spark.executor.cores","8")
conf.set("spark.python.worker.memory","30g")
conf.set("spark.driver.maxResultSize","0")

sc = SparkContext(conf=conf)

Результат

16/02/12 18:37:47 INFO SparkContext: Running Spark version 1.6.0
16/02/12 18:37:47 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
16/02/12 18:37:48 ERROR Shell: Failed to locate the winutils binary in the hadoop binary path
java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.
    at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:355)
    at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:370)
    at org.apache.hadoop.util.Shell.<clinit>(Shell.java:363)
    at org.apache.hadoop.util.StringUtils.<clinit>(StringUtils.java:79)
    at org.apache.hadoop.security.Groups.parseStaticMapping(Groups.java:104)
    at org.apache.hadoop.security.Groups.<init>(Groups.java:86)
    at org.apache.hadoop.security.Groups.<init>(Groups.java:66)
    at org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:280)
    at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:271)
    at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:248)
    at org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:763)
    at org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:748)
    at org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:621)
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2136)
    at org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2136)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.util.Utils$.getCurrentUserName(Utils.scala:2136)
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:322)
    at org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:59)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source)
    at java.lang.reflect.Constructor.newInstance(Unknown Source)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
    at py4j.Gateway.invoke(Gateway.java:214)
    at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
    at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
    at py4j.GatewayConnection.run(GatewayConnection.java:209)
    at java.lang.Thread.run(Unknown Source)
16/02/12 18:37:48 INFO SecurityManager: Changing view acls to: rmalveslocal
16/02/12 18:37:48 INFO SecurityManager: Changing modify acls to: rmalveslocal
16/02/12 18:37:48 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(rmalveslocal); users with modify permissions: Set(rmalveslocal)
16/02/12 18:37:48 INFO Utils: Successfully started service 'sparkDriver' on port 64280.
16/02/12 18:37:49 INFO Slf4jLogger: Slf4jLogger started
16/02/12 18:37:49 INFO Remoting: Starting remoting
16/02/12 18:37:49 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://[email protected]:64293]
16/02/12 18:37:49 INFO Utils: Successfully started service 'sparkDriverActorSystem' on port 64293.
16/02/12 18:37:49 INFO SparkEnv: Registering MapOutputTracker
16/02/12 18:37:49 INFO SparkEnv: Registering BlockManagerMaster
16/02/12 18:37:49 INFO DiskBlockManager: Created local directory at C:\Users\rmalveslocal\AppData\Local\Temp\1\blockmgr-4bd2f97f-8b4d-423d-a4e3-06f08ecdeca9
16/02/12 18:37:49 INFO MemoryStore: MemoryStore started with capacity 511.1 MB
16/02/12 18:37:49 INFO SparkEnv: Registering OutputCommitCoordinator
16/02/12 18:37:50 INFO Utils: Successfully started service 'SparkUI' on port 4040.
16/02/12 18:37:50 INFO SparkUI: Started SparkUI at http://10.10.5.105:4040
16/02/12 18:37:50 INFO Executor: Starting executor ID driver on host localhost
16/02/12 18:37:50 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 64330.
16/02/12 18:37:50 INFO NettyBlockTransferService: Server created on 64330
16/02/12 18:37:50 INFO BlockManagerMaster: Trying to register BlockManager
16/02/12 18:37:50 INFO BlockManagerMasterEndpoint: Registering block manager localhost:64330 with 511.1 MB RAM, BlockManagerId(driver, localhost, 64330)
16/02/12 18:37:50 INFO BlockManagerMaster: Registered BlockManager

person Kardu    schedule 11.02.2016    source источник


Ответы (1)


Вы не указали режим работы (автономный, YARN, Mesos), который вы используете, но я предполагаю, что вы используете автономный режим (для одного сервера)

Здесь играют три концепции

  • Рабочий узел — хост, на котором работает один или несколько исполнителей.
  • Executor — контейнер, в котором размещаются задачи
  • Задачи - единица работы, которая выполняется в исполнителе (части этапов, которые вместе образуют задание - оба эти термина не важны для данного обсуждения)

По умолчанию в автономном режиме все доступные ядра выделяются исполнителю. В вашем случае вы также устанавливаете его на 8, что равняется всем вашим ядрам. В результате у вас есть один исполнитель, который использует все ядра, и, поскольку вы также установили память исполнителя на 40 ГБ, вы используете только часть своей памяти для ti (40/240).

Вы можете либо увеличить объем памяти для исполнителя, чтобы позволить большему количеству задач выполняться параллельно (и иметь больше памяти для каждой), либо установить количество ядер равным 1, чтобы вы могли разместить 8 исполнителей (в этом случае вы бы вероятно, хотите установить память на меньшее число, так как 8 * 40 = 320)

person Arnon Rotem-Gal-Oz    schedule 11.02.2016
comment
Спасибо. Моя идея заключалась в том, чтобы 1 ядро ​​для каждого рабочего и 2 исполнителя внутри рабочего узла, но если я правильно понял, я был неправ, и лучшая комбинация это: conf.set(spark.executor.cores,1), conf.set(spark. num.executors,8), conf.set(spark.executor.memory, 40g). Я прав? - person Kardu; 11.02.2016
comment
8 * 40 слишком много (320, а у вас только 240), вы, вероятно, захотите spark.exeutor.cores = 8 и spark.executor.memory = 180/190G (вы получите параллелизм от 8 задач, работающих в одном и том же исполнителе и менее накладные расходы на межпроцессное взаимодействие) вам нужно оставить память для ОС и для python (вы устанавливаете это на 40G) - person Arnon Rotem-Gal-Oz; 11.02.2016
comment
@ Арнон Ротем-Гал-Оз, спасибо, и потому что мой spark.num.executors равен 8, верно? потому что у меня 8 ядер и 8 исполнителей, верно? (один исполнитель, использующий все ядро) - person Kardu; 11.02.2016
comment
количество ядер может быть 8 (или пустое, так как оно использует все по умолчанию в автономном режиме) я не помню конфиг num.executors, но даже если он есть, это не повлияет здесь - у вас есть тот, который использует всю машину - person Arnon Rotem-Gal-Oz; 11.02.2016
comment
Я меняю конфигурацию, но у меня все еще есть та же проблема, и моя память не изменилась (511,1 МБ, как моя цифра выше в вопросе). Почему меняю память на 2Gb а искра не распознается? - person Kardu; 12.02.2016
comment
2G? вы должны установить его выше, поскольку, если вы не работаете с несколькими исполнителями, он будет разделен рабочими, а также, поскольку вы используете python, вы должны оставить для этого часть памяти. - person Arnon Rotem-Gal-Oz; 12.02.2016
comment
также см. этот вопрос - person Arnon Rotem-Gal-Oz; 12.02.2016
comment
Спасибо за ваше терпение @Arnon Rotem-Gal-Oz, но я действительно не понимаю, почему я меняю конфигурацию искры (теперь у меня есть conf.set(spark.executor.memory, 25g), conf.set(spark.driver. memory, 10g), conf.set(spark.executor.cores,8) и conf.set(spark.python.worker.memory,30g), но у меня все еще есть 1 исполнитель и 511,1 МБ памяти для хранения, как на картинке выше. (пробую поменять ядра executor на 4 и 16 и все равно) :/ - person Kardu; 12.02.2016
comment
возможно, вам следует обновить вопрос, указав более подробную информацию о том, как вы запускаете Spark. - person Arnon Rotem-Gal-Oz; 12.02.2016
comment
похоже, вы настроили свою IDE для запуска spark в локальном режиме (-master local[4] или что-то подобное), это означает, что все размещено в драйвере, и все конфигурации бессмысленны (поскольку он уже установлен) - person Arnon Rotem-Gal-Oz; 12.02.2016
comment
из-за этой ошибки? Не удалось найти двоичный файл winutils в пути к двоичному файлу Hadoop? спасибо @Арнон Ротем-Гал-Оз - person Kardu; 12.02.2016