Соединение reset с помощью одноранговой сети при запуске Apache Spark Job

У нас есть две установки кластера HDP, чтобы назвать их A и B.

CLUSTER A NODES:

Он содержит в общей сложности 20 товарных машин.
Имеется 20 узлов данных.
При настройке наменовода HA существует один активный и один резервный namenode.

УКЛАДКИ КЛАСТЕРА B:

Он содержит в общей сложности 5 товарных машин.
Есть 5 datanodes.
Конфигурация HA отсутствует, и этот кластер имеет один первичный и один вторичный namenode.

У нас есть три основных компонента в нашем приложении, которые выполняют операции ETL (Извлечь, Преобразовать и Загрузить) для входящих файлов. Я буду ссылаться на эти компоненты как на E, T и L соответственно.

ТЕХНИЧЕСКИЕ ХАРАКТЕРИСТИКИ КОМПОНЕНТОВ:

Этот компонент является Apache Spark Job и работает только с кластером B.
Задача - собрать файлы из хранилища NAS и поместить их в HDFS в кластере B.

ТЕХНИЧЕСКИЕ ХАРАКТЕРИСТИКИ КОМПОНЕНТОВ:

Этот компонент также является Apache Spark Job и работает в кластере B.
Задача - собрать файлы в HDFS, записанные компонентом E, преобразовать их, а затем записать преобразованные файлы в HDFS в кластере A.

КОМПОНЕНТНЫЕ ХАРАКТЕРИСТИКИ:

Этот компонент также является задачей Apache Spark, и он работает только на кластере A.
Задача - собрать файлы, написанные компонентом T, и загрузить данные в таблицы Hive, присутствующие в кластере A.

Компонент L - это драгоценный камень среди всех трех компонентов, и мы не сталкиваемся с каким-либо сбоем в нем. В компоненте E были незначительные необъяснимые глюки, но компонент T является наиболее хлопотным.

Компонент E и T используют клиент DFS для связи с namenode.

Ниже приведен фрагмент исключения, который мы наблюдали с перерывами во время работы компонента T:

clusterA.namenode.com/10.141.160.141:8020. Trying to fail over immediately.
java.io.IOException: Failed on local exception: java.io.IOException: Connection reset by peer; Host Details : local host is: "clusterB.datanode.com"; destination host is: "clusterA.namenode.com":8020;
            at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:782)
            at org.apache.hadoop.ipc.Client.call(Client.java:1459)
            at org.apache.hadoop.ipc.Client.call(Client.java:1392)
            at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:229)
            at com.sun.proxy.$Proxy15.complete(Unknown Source)
            at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.complete(ClientNamenodeProtocolTranslatorPB.java:464)
            at sun.reflect.GeneratedMethodAccessor1240.invoke(Unknown Source)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
            at java.lang.reflect.Method.invoke(Method.java:498)
            at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:258)
            at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:104)
            at com.sun.proxy.$Proxy16.complete(Unknown Source)
            at org.apache.hadoop.hdfs.DFSOutputStream.completeFile(DFSOutputStream.java:2361)
            at org.apache.hadoop.hdfs.DFSOutputStream.closeImpl(DFSOutputStream.java:2338)
            at org.apache.hadoop.hdfs.DFSOutputStream.close(DFSOutputStream.java:2303)
            at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:72)
            at org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:106)
            at org.apache.hadoop.io.compress.CompressorStream.close(CompressorStream.java:109)
            at sun.nio.cs.StreamEncoder.implClose(StreamEncoder.java:320)
            at sun.nio.cs.StreamEncoder.close(StreamEncoder.java:149)
            at java.io.OutputStreamWriter.close(OutputStreamWriter.java:233)
            at com.abc.xyz.io.CounterWriter.close(CounterWriter.java:34)
            at com.abc.xyz.common.io.PathDataSink.close(PathDataSink.java:47)
            at com.abc.xyz.diamond.parse.map.node.AbstractOutputNode.finalise(AbstractOutputNode.java:142)
            at com.abc.xyz.diamond.parse.map.application.spark.node.SparkOutputNode.finalise(SparkOutputNode.java:239)
            at com.abc.xyz.diamond.parse.map.DiamondMapper.onParseComplete(DiamondMapper.java:1072)
            at com.abc.xyz.diamond.parse.decode.decoder.DiamondDecoder.parse(DiamondDecoder.java:956)
            at com.abc.xyz.parsing.functions.ProcessorWrapper.process(ProcessorWrapper.java:96)
            at com.abc.xyz.parser.FlumeEvent2AvroBytes.call(FlumeEvent2AvroBytes.java:131)
            at com.abc.xyz.parser.FlumeEvent2AvroBytes.call(FlumeEvent2AvroBytes.java:45)
            at org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1.apply(JavaRDDLike.scala:129)
            at org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1.apply(JavaRDDLike.scala:129)
            at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
            at scala.collection.Iterator$$anon$14.hasNext(Iterator.scala:388)
            at scala.collection.convert.Wrappers$IteratorWrapper.hasNext(Wrappers.scala:29)
            at com.abc.xyz.zzz.ParseFrameHolder$ToKafkaStream.call(ParseFrameHolder.java:123)
            at com.abc.xyz.zzz.ParseFrameHolder$ToKafkaStream.call(ParseFrameHolder.java:82)
            at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:225)
            at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:225)
            at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$35.apply(RDD.scala:927)
            at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$35.apply(RDD.scala:927)
            at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1882)
            at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1882)
            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: Connection reset by peer
            at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
            at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
            at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
            at sun.nio.ch.IOUtil.read(IOUtil.java:197)
            at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
            at org.apache.hadoop.net.SocketInputStream$Reader.performIO(SocketInputStream.java:57)
            at org.apache.hadoop.net.SocketIOWithTimeout.doIO(SocketIOWithTimeout.java:142)
            at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:161)
            at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:131)
            at java.io.FilterInputStream.read(FilterInputStream.java:133)
            at java.io.FilterInputStream.read(FilterInputStream.java:133)
            at org.apache.hadoop.ipc.Client$Connection$PingInputStream.read(Client.java:554)
            at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
            at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
            at java.io.DataInputStream.readInt(DataInputStream.java:387)
            at org.apache.hadoop.ipc.Client$Connection.receiveRpcResponse(Client.java:1116)
            at org.apache.hadoop.ipc.Client$Connection.run(Client.java:1011)

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

РЕШЕНИЯ, КОТОРЫЕ МЫ ПОТЕРЯЛИ:

Наш первый подозреваемый заключался в том, что мы перегружаем активный namenode в кластере A, так как компонент T открывает много клиентов DFS параллельно и выполняет операции с файлами по разным файлам (без проблем с обоими файлами). В наших усилиях по решению этой проблемы мы рассмотрели два ключевых параметра для namenode dfs.namenode.handler.count и ipc.server.listen.queue.size и столкнулись последний от 128 (по умолчанию) до 1024.
К сожалению, проблема все еще сохранялась в компоненте T. Мы начали использовать другой подход к проблеме. Мы сосредоточились исключительно на поиске причины возникновения соединения Reset By Peer. Согласно большому количеству обсуждений статей и стека, проблема описывается следующим образом, флаг RST был установлен одноранговым узлом, что приводит к немедленному прекращению соединения. В нашем случае мы определили, что одноранговым узлом является наменода кластера A.
Сохраняя флаг RST, я глубоко погрузился в понимание внутренних связей TCP-связи только w.r.t. причина флага RST.
Каждый сокет в дистрибутивах Linux (а не BSD) имеет две связанные с ним очереди, а именно: очередь принятия и отставания.
Во время процесса установления связи TCP все запросы хранятся в очереди задержек, пока не будут получены пакеты ACK из node, которые начали устанавливать соединение. После приема запрос передается в очередь приема, и приложение, открывшее сокет, может начать прием пакетов от удаленного клиента.
Размер очереди задержек контролируется двумя параметрами уровня ядра: net.ipv4.tcp_max_syn_backlog и net.core.somaxconn, тогда как приложение (namenode в нашем случае ) может запросить ядро для размера очереди, которое он хочет ограничить верхней границей (мы считаем, что размер очереди приема - это размер очереди, определяемый ipc.server.listen.queue.size).
Также следует отметить еще одну интересную вещь: если размер net.ipv4.tcp_max_syn_backlog больше, чем net.core.somaxconn, тогда значение прежний усечен до последнего. Эта претензия основана на документации Linux и может быть найдена в https://linux.die.net/man/2/listen.
Возвращаясь к точке, когда задержка заполняется полностью, TCP ведет себя двумя способами, и это поведение также можно контролировать с помощью параметра ядра net.ipv4.tcp_abort_on_overflow. По умолчанию оно установлено в 0 и заставляет ядро удалять любые новые SYN-пакеты, когда заполнение заполнено, что, в свою очередь, позволяет отправителю повторно отправить SYN-пакеты. Если установлено значение 1, ядро отметит флаг RST в пакете и отправит его отправителю, что резко прекратит соединение.
Мы проверили значение вышеупомянутых параметров ядра и выяснили, что для параметра net.core.somaxconn установлено значение 1024, net.ipv4.tcp_abort_on_overflow установлено значение 0 и net.ipv4.tcp_max_syn_backlog установлено на 4096 по всем машинам в обоих кластерах.
Единственный подозреваемый, который мы оставили сейчас, это коммутаторы, которые подключают кластер A к кластеру B, потому что ни одна из машин в любом из кластеров никогда не установит флаг RST в качестве параметра net.ipv4. tcp_abort_on_overflow установлено в 0.

МОИ ВОПРОСЫ

Из документации HDFS видно, что DFS Client использует RPC для связи с namenode для выполнения файловых операций. Каждый вызов RPC включает установление TCP-соединения с namenode?
Определяет ли параметр ipc.server.listen.queue.size длину очереди приема сокета, в которой namenode принимает запросы RPC?
Может ли namenode неявно закрыть соединения с клиентом DFS, когда он находится под большой нагрузкой, что позволяет ядру отправить пакет с установленным флажком RST, даже если для параметра kernel net.ipv4.tcp_abort_on_overflow установлено значение 0?
Переключатели L2 или L3 (используемые для подключения компьютеров в наших двух кластерах), способные устанавливать флаг RST, потому что они не способны обрабатывать пакетные трафики?

Наш следующий подход к этой проблеме заключается в том, чтобы определить, какая машина или коммутатор (нет задействованного маршрутизатора) устанавливает флаг RST, анализируя пакеты с помощью tcpdump или wireshark. Мы также увеличим размер всех упомянутых выше очередей до 4096, чтобы эффективно обрабатывать пакетный трафик.

В журналах номенклада нет никаких признаков каких-либо исключений, за исключением того, что загрузка соединения Namenode, как видно из Ambari, просматривалась в определенные моменты времени и не обязательно, когда произошло событие Connection Reset By Peer.

В заключение я хотел знать, направляется ли мы на правильный путь, чтобы решить эту проблему, или мы просто ударим в тупик?

P.S. Приносим извинения за длину содержимого в моем вопросе. Я хотел представить весь контекст читателям, прежде чем обращаться за помощью или предложениями. Спасибо за терпеливость.

Ответ 1