Verbindung durch Peer zurückgesetzt, während Apache Spark Job ausgeführt wird

Wir haben zwei HDP-Cluster-Setups, nennen wir sie A und B.

CLUSTER A NODES :

• Es enthält insgesamt 20 Warenautomaten.
• Es gibt 20 Datenknoten.
• Wenn namenode HA konfiguriert ist, gibt es einen aktiven und einen Standby-namen.

CLUSTER B NODES :

• Es enthält insgesamt 5 Warenautomaten.
• Es gibt 5 Datanodes.
• Es ist kein HA konfiguriert und dieser Cluster hat einen primären und einen sekundären Namen.

Wir haben drei Hauptkomponenten in unserer Anwendung, die einen ETL-Vorgang (Extrahieren, Transformieren und Laden) für eingehende Dateien durchführen. Ich werde diese Komponenten als E, T bzw. L bezeichnen.

EIGENSCHAFTEN DER KOMPONENTE :

• Diese Komponente ist ein Apache Spark Job und läuft ausschließlich auf Cluster B.
• Es ist Aufgabe, Dateien von einem NAS-Speicher abzuholen und sie in Cluster B in HDFS zu speichern.

KOMPONENTEN-T-MERKMALE :

• Diese Komponente ist auch ein Apache Spark Job und läuft auf Cluster B.
• Die Aufgabe besteht darin, die Dateien in HDFS, die von der Komponente E geschrieben wurden, aufzunehmen, umzuwandeln und dann die transformierten Dateien in Cluster A in HDFS zu schreiben.

KOMPONENTE L-MERKMALE :

• Diese Komponente ist ebenfalls ein Apache Spark-Job und läuft ausschließlich auf Cluster A.
• Es ist Aufgabe, die von Komponente T geschriebenen Dateien zu übernehmen und die Daten in Hive-Tabellen zu laden, die in Cluster A vorhanden sind.

Die Komponente L ist das Juwel unter allen drei Komponenten und wir haben keine Störungen darin gesehen. In der Komponente E gab es geringfügige ungeklärte Störungen, aber die Komponente T ist die schwierigste.

Die Komponenten E und T verwenden beide den DFS-Client, um mit dem Namen zu kommunizieren.

Nachfolgend finden Sie einen Auszug der Ausnahme, die wir beim Ausführen der Komponente T intermittierend beobachtet haben:

Wie bereits erwähnt, sehen wir uns dieser Ausnahme sehr zeitweise gegenüber. Wenn sie auftritt, bleibt unsere Anwendung stecken und veranlasst uns, sie neu zu starten.

LÖSUNGEN, DIE WIR VERSUCHTEN:

• Unser erster Verdacht war, dass wir den aktiven namenode in Cluster A überlasten, da Komponente T viele DFS-Clients parallel öffnet und Dateioperationen für verschiedene Dateien durchführt (kein Konflikt in denselben Dateien). In unserem Bemühen, dieses Problem anzugehen, haben wir uns zwei Schlüsselparameter für den namenode dfs.namenode.handler.count und ipc.server.listen.queue.size angesehen Letzteres von 128 (Standard) bis 1024.

• Leider blieb das Problem in Komponente T bestehen. Wir begannen einen anderen Ansatz für das Problem. Wir haben uns ausschließlich darauf konzentriert, den Grund für das Auftreten von Connection Reset By Peer zu finden. Laut einer Vielzahl von Artikeln und Diskussionen zum Austausch von Stacks wird das Problem wie folgt beschrieben: Das Flag RST wurde vom Peer gesetzt, was zu einem sofortigen Abbruch der Verbindung führt . In unserem Fall haben wir festgestellt, dass der Peer der Name von Cluster A ist.

• Ich behalte die RST-Flagge im Hinterkopf, und habe mich intensiv damit beschäftigt, die Interna der TCP-Kommunikation nur zu verstehen. der Grund des RST Flags.

• Jeder Socket in Linux-Distributionen (nicht BSD) hat zwei Warteschlangen, nämlich die accept- und die backlog-Queue.
• Während des TCP-Handshake-Prozesses werden alle Anforderungen in der Backlog-Warteschlange gehalten, bis ACK-Pakete von dem Knoten empfangen werden, der mit dem Aufbau der Verbindung begonnen hat. Nach dem Empfang wird die Anfrage in die Annahmeliste übertragen und die Anwendung, die den Socket geöffnet hat, kann Pakete vom Remote-Client empfangen.
• Die Größe der Backlog-Queue wird durch zwei Kernel-Level-Parameter gesteuert, nämlich net.ipv4.tcp_max_syn_backlog und net.core.somaconn , während die Anwendung (narnode in unserem Fall) ) kann den Kernel für die von ihm gewünschte Warteschlangengröße anfordern, die durch eine obere Grenze begrenzt ist (wir glauben, dass die akzeptierte Warteschlangengröße die durch ipc.server.listen.queue.size definierte Warteschlangengröße ist).
• Außerdem ist hier noch eine interessante Sache zu beachten: Wenn die Größe von net.ipv4.tcp_max_syn_backlog größer ist als net.core.somaconn , dann ist der Wert von Ersterer wird auf den Letzteren verkürzt. Dieser Anspruch basiert auf der Linux-Dokumentation und kann unter Ссылка gefunden werden.

• Kommen wir zu dem Punkt zurück, an dem der Backlog vollständig gefüllt ist. TCP verhält sich auf zwei Arten und dieses Verhalten kann auch durch einen Kernel-Parameter namens net.ipv4.tcp_abort_on_overflow gesteuert werden. Dies ist standardmäßig auf 0 gesetzt und bewirkt, dass der Kernel alle neuen SYN-Pakete löscht, wenn der Rückstand voll ist, wodurch der Absender wiederum SYN-Pakete senden kann.Wenn der Kernel auf 1 gesetzt ist, markiert er das RST-Flag in einem Paket und sendet es an den Sender, wodurch die Verbindung abrupt beendet wird.

• Wir haben den Wert der oben genannten Kernel-Parameter überprüft und festgestellt, dass net.core.somaconn auf 1024 gesetzt ist, net.ipv4.tcp_abort_on_overflow ist 0 und net.ipv4.tcp_max_syn_backlog wird auf allen Computern in beiden Clustern auf 4096 festgelegt.

• Der einzige Verdacht, den wir jetzt noch haben, sind die Switches, die Cluster A mit Cluster B verbinden, weil keiner der Maschinen in irgendeinem Cluster das RST-Flag als Parameter net.ipv4 setzen wird. tcp_abort_on_overflow ist auf 0 gesetzt.

MEINE FRAGEN

• Aus der HDFS-Dokumentation geht hervor, dass der DFS-Client RPC für die Kommunikation mit dem namenode zum Ausführen von Dateioperationen verwendet. Beinhaltet jeder RPC-Aufruf den Aufbau einer TCP-Verbindung zu Namenknoten?
• Definiert der Parameter ipc.server.listen.queue.size die Länge der Annahmewarteschlange des Sockets, an der nomenode RPC-Anfragen akzeptiert?
• Kann der namenode bei hoher Auslastung implizit Verbindungen zum DFS-Client schließen, so dass der Kernel ein Paket mit gesetztem RST-Flag sendet, selbst wenn der Kernel-Parameter net.ipv4.tcp_abort_on_overflow auf gesetzt ist 0?
• Sind L2- oder L3-Switches (zum Verbinden der Maschinen in unseren beiden Clustern) in der Lage, das RST-Flag zu setzen, weil sie nicht in der Lage sind, stoßartiges Datenverkehr zu verarbeiten?

Unser nächster Ansatz für dieses Problem besteht darin, zu ermitteln, welche Maschine oder welcher Switch (es gibt keinen Router) das RST-Flag durch Analyse der Pakete mit tcpdump oder wireshark setzt. Wir werden auch die Größe aller oben erwähnten Warteschlangen auf 4096 erhöhen, um stoßartigen Verkehr wirksam zu handhaben.

In den namenode-Protokollen sind keine Ausnahmen zu sehen, außer dass die Namendeode-Verbindung in Ambari zu bestimmten Zeitpunkten und nicht unbedingt bei der Ausnahme "Connection Reset by Peer" angezeigt wurde.

Zum Schluss wollte ich wissen, ob wir auf dem richtigen Weg sind, um dieses Problem zu lösen, oder werden wir gerade in eine Sackgasse geraten?

P.S. Ich entschuldige mich für die Länge des Inhalts in meiner Frage. Ich wollte den Lesern den gesamten Kontext vorstellen, bevor ich um Hilfe oder Anregungen bat. Vielen Dank für Ihre Geduld.