Сбой приложения Dask, когда слишком много работников терпят неудачу - PullRequest
Новая
       9

Сбой приложения Dask, когда слишком много работников терпят неудачу

0 голосов
/ 21 июня 2019

Я запускаю приложение Dask (1.2), используя Dask YARN (0.6.0) в кластере EMR.Сегодня я попал в ситуацию, когда мои работники выходили из строя (из-за ошибки HDFS) и мотка. ApplicationMaster постоянно воссоздает новых работников.Есть ли способ дать указание Dask YARN отменить заявку, если слишком много работников потерпят неудачу?

В частности, мои журналы Application Master выглядят так: 

19/06/21 16:00:27 INFO skein.ApplicationMaster: RESTARTING: adding new container to replace dask.worker_805.
19/06/21 16:00:27 INFO skein.ApplicationMaster: REQUESTED: dask.worker_806
19/06/21 16:00:27 WARN skein.ApplicationMaster: FAILED: dask.worker_804 - Could not obtain block: BP-1234110000-10.174.17.184-1561122672601:blk_1073741831_1007 file=/user/hadoop/.skein/application_1561122685021_0003/FED3ABF369AAE224B4BB8A3A77120E1C/cached_volume.sqlite3
org.apache.hadoop.hdfs.BlockMissingException: Could not obtain block: BP-1234110000-10.174.17.184-1561122672601:blk_1073741831_1007 file=/user/hadoop/.skein/application_1561122685021_0003/FED3ABF369AAE224B4BB8A3A77120E1C/cached_volume.sqlite3
    at org.apache.hadoop.hdfs.DFSInputStream.chooseDataNode(DFSInputStream.java:983)
    at org.apache.hadoop.hdfs.DFSInputStream.blockSeekTo(DFSInputStream.java:642)
    at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSInputStream.java:882)
    at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:934)
    at java.io.DataInputStream.read(DataInputStream.java:100)
    at org.apache.hadoop.io.IOUtils.copyBytes(IOUtils.java:85)
    at org.apache.hadoop.io.IOUtils.copyBytes(IOUtils.java:59)
    at org.apache.hadoop.io.IOUtils.copyBytes(IOUtils.java:119)
    at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:366)
    at org.apache.hadoop.yarn.util.FSDownload.copy(FSDownload.java:267)
    at org.apache.hadoop.yarn.util.FSDownload.access$000(FSDownload.java:63)
    at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:361)
    at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:359)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:422)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1698)
    at org.apache.hadoop.yarn.util.FSDownload.call(FSDownload.java:358)
    at org.apache.hadoop.yarn.util.FSDownload.call(FSDownload.java:62)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

до бесконечности

1 Ответ

1 голос
/ 25 июня 2019

Если вы используете главный конструктор, вы можете установить максимальное количество рабочих перезапусков с помощью worker_restarts kwarg: 

# Allow a maximum of 3 worker restarts before failure
cluster = YarnCluster(worker_restarts=3, ...)

В качестве альтернативы, если вы используете пользовательскую спецификацию , вы можете указатьмаксимальное количество разрешенных перезапусков с max_restarts . 

# /path/to/spec.yaml
name: dask
queue: myqueue

services:
  dask.worker:
    # Don't start any workers initially
    instances: 0
    # A maximum of 3 worker failures are allowed before failure
    max_restarts: 3
    # Restrict workers to 4 GiB and 2 cores each
    resources:
      memory: 4 GiB
      vcores: 2
    # Distribute this python environment to every worker node
    files:
      environment: /path/to/my/environment.tar.gz
    # The bash script to start the worker
    # Here we activate the environment, then start the worker
    script: |
      source environment/bin/activate
      dask-yarn services worker
...