Почему приложение Spark в YARN не работает с FetchFailedException из-за отказа Connection?

Question

Я пользуюсь spark version 1.6.3 и yarn version 2.7.1.2.3 прилагается HDP-2.3.0.0-2557. Becuase, искровая версия слишком устарела в версии HDP, которую я использую, я предпочитаю использовать другую искру в качестве режима пряжи удаленно.

Вот как я запускаю искровую оболочку;

./spark-shell --master yarn-client 

Все, кажется, прекрасно, sparkContext инициализации sqlContext инициализируется. Я даже могу получить доступ к моим таблицам. Но в некоторых случаях у него возникают проблемы, когда он пытается подключиться к блочным менеджерам.

Я не эксперт, но я думаю, что администраторы блоков, пока я запускаю его в режиме пряжи, работают на моем кластере пряжи. Мне впервые показалась сетевая проблема, и я не хотел спрашивать ее здесь. Но это случается в некоторых случаях, которые я еще не мог понять. Поэтому мне кажется, что это не проблема сети.

Вот код;

def df = sqlContext.sql("select * from city_table") 

Коды ниже хорошо работают;

df.limit(10).count() 

Но размер более 10, я не знаю, это изменяется при каждом запуске;

df.count() 

Это вызывает исключение;

6/12/30 07:31:04 INFO MapOutputTrackerMaster: Size of output statuses for shuffle 2 is 157 bytes 
16/12/30 07:31:19 WARN TaskSetManager: Lost task 0.0 in stage 5.0 (TID 8, 172.27.247.204): FetchFailed(BlockManagerId(2, 172.27.247.204, 56093), shuffleId=2, mapId=0, reduceId=0, message= 
org.apache.spark.shuffle.FetchFailedException: Failed to connect to /172.27.247.204:56093 
    at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:323) 
    at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:300) 
    at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:51) 
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) 
    at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) 
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) 
    at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:32) 
    at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39) 
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) 
    at org.apache.spark.sql.execution.aggregate.TungstenAggregationIterator.processInputs(TungstenAggregationIterator.scala:504) 
    at org.apache.spark.sql.execution.aggregate.TungstenAggregationIterator.<init>(TungstenAggregationIterator.scala:686) 
    at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1$$anonfun$2.apply(TungstenAggregate.scala:95) 
    at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1$$anonfun$2.apply(TungstenAggregate.scala:86) 
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710) 
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$20.apply(RDD.scala:710) 
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270) 
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270) 
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
    at org.apache.spark.scheduler.Task.run(Task.scala:89) 
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:745) 
Caused by: java.io.IOException: Failed to connect to /172.27.247.204:56093 
    at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:216) 
    at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:167) 
    at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:90) 
    at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:140) 
    at org.apache.spark.network.shuffle.RetryingBlockFetcher.access$200(RetryingBlockFetcher.java:43) 
    at org.apache.spark.network.shuffle.RetryingBlockFetcher$1.run(RetryingBlockFetcher.java:170) 
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
    ... 3 more 
Caused by: java.net.ConnectException: Connection refused: /172.27.247.204:56093 
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) 
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) 
    at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:224) 
    at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:289) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:528) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468) 
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382) 
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354) 
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111) 
    ... 1 more 

) 

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

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

спасибо.

Отредактировано:

Я вижу что-то в журналах;

17/01/02 06:45:17 INFO DAGScheduler: Executor lost: 2 (epoch 13) 
17/01/02 06:45:17 INFO BlockManagerMasterEndpoint: Trying to remove executor 2 from BlockManagerMaster. 
17/01/02 06:45:17 INFO BlockManagerMasterEndpoint: Removing block manager BlockManagerId(2, 172.27.247.204, 51809) 
17/01/02 06:45:17 INFO BlockManagerMaster: Removed 2 successfully in removeExecutor 
17/01/02 06:45:17 INFO YarnScheduler: Removed TaskSet 3.0, whose tasks have all completed, from pool 
17/01/02 06:45:24 INFO BlockManagerMasterEndpoint: Registering block manager 172.27.247.204:51809 with 511.1 MB RAM, BlockManagerId(2, 172.27.247.204, 51809) 

Иногда, повторение его на другой блок менеджер работает, но, поскольку максимально допустимое количество раз, которое является 4 в превышении по умолчанию, он никогда не заканчивается большую часть времени.

Edited 2:

Пряжа действительно очень молчит об этом, но я думаю, что это сеть проблема, я мог бы итерацию проблемы куда-то;

Эта искра развернута за пределами среды HDP. Когда искра подает заявку на пряжу, пряжа сообщает искровому драйверу о менеджере блоков и исполнителях. Исполнители - это узлы данных в кластере HDP и имеют разные IP-адреса в своей частной сети. Но когда дело доходит до информирования искрового драйвера за пределами кластера, он дает одинаковый и всегда единый IP для всех исполнителей. Это связано с тем, что все узлы в HDP-кластере выходят за маршрутизатор и с тем же IP-адресом. Предположим, что IP-адрес равен 150.150.150.150, когда искровому драйверу необходимо подключиться и спросить что-нибудь от этих исполнителей, он пытается использовать этот IP-адрес. Но этот IP-адрес фактически является внешним IP-адресом всего кластера, а не отдельным IP-узлом данных.

Есть ли способ сделать пряжу информировать об исполнителях (блочных менеджерах) со своим личным ip. Потому что их частные IP-адреса также доступны с компьютера, на котором работает этот искровой драйвер.

Answer 1

FetchFailedException исключение выбрасывается, когда задача редуктора (для ShuffleDependency) не могла получить блоки перетасовки. Это обычно означает, что исполнитель (с BlockManager для перетасовки блоков) умер, и, следовательно, исключение:

Caused by: java.io.IOException: Failed to connect to /172.27.247.204:56093 

Исполнитель может OOMed (= OutOfMemoryError выброшен) или ПРЯЖА решила убить его из-за чрезмерное использование памяти.

Вы должны просмотреть журналы приложения Spark с помощью команды yarn logs и выяснить основную причину проблемы.

yarn logs -applicationId <application ID> [options] 

Вы также можете просмотреть статус исполнителей приложения Spark на вкладке «Исполнители» в веб-интерфейсе.

Искры обычно восстанавливаются с FetchFailedException путем повторного запуска затронутых задач. Используйте веб-интерфейс, чтобы узнать, как работает приложение Spark. FetchFailedException может быть связано с временной памятью «икота».