YARN nie wywiązuje się z zasobów w oparciu o udziały w wartości godziwej podczas uruchamiania zadania Spark

Question

Mam problem z ponownym równoważeniem zasobów zadań Apache Spark w kolejkach YARN Fair Scheduled.

Dla testów skonfigurowałem Hadoop 2.6 (wypróbowałem również 2.7), aby uruchomić tryb pseudo-rozproszony z lokalnym HDFS na MacOS. Do przesłania pracy wykorzystano wersję "Pre-build Spark 1.4 dla Hadoop 2.6 i późniejszych" (wypróbowano także 1.5) z Spark's website.

Podczas testowania z podstawową konfiguracją zadań Hadoop MapReduce program Fair Scheduler działa zgodnie z oczekiwaniami: Gdy zasoby klastra przekroczą określoną wartość maksymalną, obliczane są udziały uczciwe, a zasoby dla zadań w różnych kolejkach są wyrównywane i wyważane na podstawie tych obliczeń.

Ten sam test jest wykonywany z zadaniami Spark, w tym przypadku ZARZĄD dokona poprawnych obliczeń udziałów w targach dla każdego zadania, ale zasoby dla pojemników Spark nie są ponownie zrównoważone.

Oto moje pliki conf:

$ HADOOP_HOME/etc/Hadoop/przędza-site.xml

<?xml version="1.0" encoding="UTF-8"?> 
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?> 
<configuration> 
    <property> 
     <name>yarn.nodemanager.aux-services</name> 
     <value>mapreduce_shuffle</value> 
    </property> 
    <property> 
     <name>yarn.nodemanager.aux-services.spark_shuffle.class</name> 
     <value>org.apache.spark.network.yarn.YarnShuffleService</value> 
    </property> 
    <property> 
     <name>yarn.resourcemanager.scheduler.class</name> 
     <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value> 
    </property> 
    <property> 
     <name>yarn.scheduler.fair.preemption</name> 
     <value>true</value> 
    </property> 
</configuration> 

$ HADOOP_HOME/etc/Hadoop/fair-scheduler.xml

<?xml version="1.0" encoding="UTF-8"?> 
<allocations> 
    <defaultQueueSchedulingPolicy>fair</defaultQueueSchedulingPolicy> 
    <queue name="prod"> 
     <weight>40</weight> 
     <schedulingPolicy>fifo</schedulingPolicy> 
    </queue> 
    <queue name="dev"> 
     <weight>60</weight> 
     <queue name="eng" /> 
     <queue name="science" /> 
    </queue> 
    <queuePlacementPolicy> 
     <rule name="specified" create="false" /> 
     <rule name="primaryGroup" create="false" /> 
     <rule name="default" queue="dev.eng" /> 
    </queuePlacementPolicy> 
</allocations> 

$ HADOOP_HOME/etc/Hadoop/core-site.xml

<?xml version="1.0" encoding="UTF-8"?> 
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?> 
<configuration> 
    <property> 
     <name>fs.defaultFS</name> 
     <value>hdfs://localhost:9000</value> 
    </property> 
</configuration> 

$ HADOOP_HOME/etc/Hadoop/core-site.xml

<?xml version="1.0" encoding="UTF-8"?> 
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?> 
<configuration> 
    <property> 
     <name>dfs.replication</name> 
     <value>1</value> 
    </property> 
</configuration> 

a sprawa test jest:

Uruchom zadanie na "prod" kolejki o masie 40 (należy przeznaczyć 40% wszystkich zasoby), zgodnie z oczekiwaniami, że praca zajmuje wszystkie potrzebne wolne zasoby (62,5% zasobów klastra).

./bin/spark-submit --class org.apache.spark.examples.SparkPi \ 
--master yarn-cluster \ 
--driver-memory 512M \ 
--executor-memory 768M \ 
--executor-cores 1 \ 
--num-executors 2 \ 
--queue prod \ 
lib/spark-examples*.jar 100000 

Potem uruchomić tę samą pracę w kolejce „dev.eng” o wadze 60, który oznacza pracę musi przeznaczyć 60% wszystkich zasobów i zmniejszenie zasobów pierwszej pracy do ~ 40%.

Niestety zasoby klastra się nie zmieniają - 62,5% dla pierwszej pracy i 37,5% dla drugiej.

Answer 1

Fair Scheduler nie zabija kontenerów dla pierwszego zadania, tylko czeka, aż niektóre zasoby zostaną zwolnione i zarezerwuje je do wykorzystania przez drugie zadanie. Jeśli nie zasoby są wolne od pierwszego zadania, program planujący nie może przypisać tych zasobów do drugiego zadania.

W zadaniach MapReduce każde zadanie mapy lub zadania zmniejszania wymaga utworzenia instancji nowego kontenera, a program planujący może zablokować zadanie, aby utworzyć nowe kontenery, jeśli przekroczyło ono swoją ofertę (na podstawie pojemności kolejki).

W Spark rzeczy są różne, executorzy są inicjowani na początku zadania i wysyłane są do nich różne zadania (etapy). Następnie zasoby nie są wolne i nie można ich ponownie rozdzielić.

Może być dynamiczna alokacja może pomóc: http://spark.apache.org/docs/1.6.1/configuration.html#dynamic-allocation

Answer 2

Musisz ustawić jeden z limitów czasu wywłaszczania w swojej xml alokacji. Jeden dla minimalnego udziału, a drugi dla sprawiedliwego podziału, oba są w sekundach. Domyślnie limity czasu nie są ustawione.

Od Hadoop: The Definitive Guide 4th Edition

Jeśli kolejka czeka tak długo, jak jej minimalnego limitu czasu wywłaszczania akcji bez otrzymania jej minimalny gwarantowany udział, to planista może uprzedzić innych pojemników. Domyślny limit czasu jest ustawiany dla wszystkich kolejek za pośrednictwem domyślnego elementu MinSharePreemptionTimeout najwyższego poziomu w pliku alokacji , a także dla każdej kolejki, ustawiając element minSharePreemptionTimeout dla kolejki w postaci .

Podobnie, jeśli kolejka pozostaje poniżej połowy swojego sprawiedliwego udziału tak długo, jak jako czas oczekiwania na prawidłowy udział, wówczas planista może przejąć inne pojemniki z wyprzedzeniem . Domyślny limit czasu jest ustawiony dla wszystkich kolejek za pomocą domyślnego elementu defaultFairSharePreemptionTimeout o wartości w kolejce alokacji , a także dla każdej kolejki, ustawiając fairSharePreemptionTimeout w kolejce. Wartość progowa może również zostać zmieniona z wartości domyślnej na 0,5 , ustawiając defaultFairSharePreemptionThreshold i fairSharePreemptionThreshold (za kolejkę).