Hallo! Wir schulen Menschen im Umgang mit großen Datenmengen. Man kann sich ein Bildungsprogramm über Big Data nicht ohne einen Cluster vorstellen, an dem alle Teilnehmer gemeinsam arbeiten. Aus diesem Grund ist er immer Teil unseres Programms 🙂 Wir kümmern uns um die Einrichtung, Feinabstimmung und Verwaltung, während die Teilnehmer dort MapReduce-Jobs ausführen und Spark verwenden.
In diesem Beitrag werden wir beschreiben, wie wir das Problem der ungleichmäßigen Clusterauslastung gelöst haben, indem wir unseren eigenen Autoscaler geschrieben haben, der die Cloud nutzt. .
Problem
Unser Cluster wird nicht ganz im typischen Modus verwendet. Die Auslastung ist stark ungleichmäßig. Zum Beispiel gibt es praktische Übungen, bei denen alle 30 Teilnehmer und der Dozent auf den Cluster zugreifen und ihn nutzen. Oder es gibt Tage vor Fristen, an denen die Auslastung stark ansteigt. Ansonsten läuft der Cluster im Modus einer Unterauslastung.
Lösung Nr. 1 – einen Cluster zu halten, der die Spitzenlasten bewältigen kann, der aber in der übrigen Zeit ungenutzt bleibt.
Lösung Nr. 2 – einen kleinen Cluster zu betreiben, in den man manuell Knoten vor den Unterrichtseinheiten und während der Spitzenlasten hinzufügt.
Lösung Nr. 3 – einen kleinen Cluster zu betreiben und einen Autoscaler zu schreiben, der die aktuelle Auslastung des Clusters überwacht und mithilfe verschiedener APIs selbstständig Knoten zum Cluster hinzufügt und entfernt.
In diesem Beitrag werden wir über Lösung Nr. 3 sprechen. Ein solcher Autoscaler hängt stark von externen Faktoren ab und nicht von internen, und Anbieter stellen ihn oft nicht zur Verfügung. Wir nutzen die Cloud-Infrastruktur von Mail.ru Cloud Solutions und haben einen Autoscaler geschrieben, der die API von MCS verwendet. Da wir Datenarbeit lehren, haben wir uns entschieden zu zeigen, wie Sie einen ähnlichen Autoscaler für Ihre Zwecke schreiben und mit Ihrer Cloud verwenden können.
Voraussetzungen
Zuerst müssen Sie einen Hadoop-Cluster haben. Wir verwenden zum Beispiel die HDP-Distribution.
Damit Knoten schnell hinzugefügt und entfernt werden können, sollten Sie eine gewisse Rollenverteilung auf den Knoten haben.
- Master-Knoten. Hier muss nicht viel erklärt werden: der Hauptknoten des Clusters, auf dem zum Beispiel der Spark-Treiber ausgeführt wird, wenn Sie den interaktiven Modus verwenden.
- Daten-Knoten. Das ist der Knoten, auf dem die Daten im HDFS gespeichert sind und auf dem auch die Berechnungen durchgeführt werden.
- Rechenknoten. Das ist ein Knoten, auf dem Sie nichts im HDFS speichern, aber Berechnungen durchgeführt werden.
Wichtiger Punkt. Das Auto-Scaling erfolgt durch Knoten des dritten Typs. Wenn Sie beginnen, Knoten des zweiten Typs zu entfernen und hinzuzufügen, wird die Reaktionsgeschwindigkeit stark sinken – Decommissioning und Recommissioning wird Stunden in Ihrem Cluster in Anspruch nehmen. Das ist natürlich nicht das, was man von Auto-Scaling erwartet. Das heißt, wir lassen die Knoten des ersten und zweiten Typs unberührt. Sie werden einen minimal funktionsfähigen Cluster bilden, der während der gesamten Programmlaufzeit existiert.
Unser Auto-Scaler ist also in Python 3 geschrieben, nutzt die Ambari API zur Verwaltung der Cluster-Services, verwendet (MCS) zum Starten und Stoppen von Maschinen.
Architektur der Lösung
- Modul
autoscaler.py. Es sind drei Klassen definiert: 1) Funktionen zur Arbeit mit Ambari, 2) Funktionen zur Arbeit mit MCS, 3) Funktionen, die direkt mit der Logik des Auto-Scalers verbunden sind. - Skript
observer.py. Besteht im Wesentlichen aus verschiedenen Regeln: Wann und in welchen Momenten die Funktionen des Auto-Scalers aufgerufen werden sollen. - Die Datei mit den Konfigurationseinstellungen
config.py. Sie enthält unter anderem eine Liste von Knoten, die für das Auto-Scaling zugelassen sind, sowie andere Parameter, die beispielsweise beeinflussen, wie lange man warten sollte, nachdem ein neuer Knoten hinzugefügt wurde. Dort befinden sich auch Zeitstempel für den Beginn der Sitzungen, damit die maximal erlaubte Konfiguration des Clusters vor der Sitzung gestartet wird.
Schauen wir uns nun die Codeausschnitte in den ersten beiden Dateien an.
1. Modul autoscaler.py
Klasse Ambari
So sieht ein Codeausschnitt aus, der die Klasse Ambari enthält.:
class Ambari:
def __init__(self, ambari_url, cluster_name, headers, auth):
self.ambari_url = ambari_url
self.cluster_name = cluster_name
self.headers = headers
self.auth = auth
def stop_all_services(self, hostname):
url = self.ambari_url + self.cluster_name + '/hosts/' + hostname + '/host_components/'
url2 = self.ambari_url + self.cluster_name + '/hosts/' + hostname
req0 = requests.get(url2, headers=self.headers, auth=self.auth)
services = req0.json()['host_components']
services_list = list(map(lambda x: x['HostRoles']['component_name'], services))
data = {
"RequestInfo": {
"context":"Stop All Host Components",
"operation_level": {
"level":"HOST",
"cluster_name": self.cluster_name,
"host_names": hostname
},
"query":"HostRoles/component_name.in({0})".format(",".join(services_list))
},
"Body": {
"HostRoles": {
"state":"INSTALLED"
}
}
}
req = requests.put(url, data=json.dumps(data), headers=self.headers, auth=self.auth)
if req.status_code in [200, 201, 202]:
message = 'Anfrage akzeptiert'
else:
message = req.status_code
return messageOben kann man die Implementierung der Funktion ansehen stop_all_services, die alle Dienste auf dem gewünschten Knoten des Clusters stoppt.
Der Klasse Ambari enthält. geben Sie Folgendes ein:
ambari_url, zum Beispiel in folgender Form'http://localhost:8080/api/v1/clusters/',cluster_name– der Name Ihres Clusters in Ambari,headers = {'X-Requested-By': 'ambari'}- und darin
authliegen Ihr Benutzername und Passwort für Ambari:auth = ('login', 'password').
Die Funktion besteht im Wesentlichen aus ein paar REST API-Anfragen an Ambari. Logisch gesehen erhalten wir zunächst die Liste der gestarteten Dienste auf dem Knoten und fordern dann an, dass diese Dienste in diesem Cluster auf diesem Knoten in den Zustand INSTALLEDversetzt werden. Die Funktionen zum Starten aller Dienste, zum Versetzen von Knoten in den Zustand Wartung usw. sehen ähnlich aus – es sind einfach mehrere Anfragen über die API.
Die Klasse Mcs
So sieht ein Codeausschnitt aus, der die Klasse Mcs:
class Mcs:
def __init__(self, id1, id2, password):
self.id1 = id1
self.id2 = id2
self.password = password
self.mcs_host = 'https://infra.mail.ru:8774/v2.1'
def vm_turn_on(self, hostname):
self.token = self.get_mcs_token()
host = self.hostname_to_vmname(hostname)
vm_id = self.get_vm_id(host)
mcs_url1 = self.mcs_host + '/servers/' + self.vm_id + '/action'
headers = {
'X-Auth-Token': '{0}'.format(self.token),
'Content-Type': 'application/json'
}
data = {'os-start' : 'null'}
mcs = requests.post(mcs_url1, data=json.dumps(data), headers=headers)
return mcs.status_codeDer Klasse Mcs wir übergeben die Projekt-ID in der Cloud und die Benutzer-ID sowie das Passwort. In der Funktion vm_turn_on Wir möchten eine der Maschinen einschalten. Die Logik hier ist etwas komplizierter. Zu Beginn des Codes wird der Aufruf von drei anderen Funktionen vorgenommen: 1) Wir müssen ein Token abrufen, 2) Wir müssen den Hostnamen in den Namen der Maschine im MCS umwandeln, 3) Die ID dieser Maschine abrufen. Danach machen wir einfach eine POST-Anfrage und starten diese Maschine.
So sieht die Funktion zum Abrufen des Tokens aus:
def get_mcs_token(self):
url = 'https://infra.mail.ru:35357/v3/auth/tokens?nocatalog'
headers = {'Content-Type': 'application/json'}
data = {
'auth': {
'identity': {
'methods': ['password'],
'password': {
'user': {
'id': self.id1,
'password': self.password
}
}
},
'scope': {
'project': {
'id': self.id2
}
}
}
}
params = (('nocatalog', ''),)
req = requests.post(url, data=json.dumps(data), headers=headers, params=params)
self.token = req.headers['X-Subject-Token']
return self.tokenDie Klasse Autoscaler
In dieser Klasse sind Funktionen enthalten, die sich auf die eigentliche Logik der Arbeit beziehen.
So sieht ein Teil des Codes dieser Klasse aus:
class Autoscaler:
def __init__(self, ambari, mcs, scaling_hosts, yarn_ram_per_node, yarn_cpu_per_node):
self.scaling_hosts = scaling_hosts
self.ambari = ambari
self.mcs = mcs
self.q_ram = deque()
self.q_cpu = deque()
self.num = 0
self.yarn_ram_per_node = yarn_ram_per_node
self.yarn_cpu_per_node = yarn_cpu_per_node
def scale_down(self, hostname):
flag1 = flag2 = flag3 = flag4 = flag5 = False
if hostname in self.scaling_hosts:
while True:
time.sleep(5)
status1 = self.ambari.decommission_nodemanager(hostname)
if status1 == 'Request accepted' or status1 == 500:
flag1 = True
logging.info('Decommission-Anfrage akzeptiert: {0}'.format(flag1))
break
while True:
time.sleep(5)
status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
if status3 == 'INSTALLED':
flag3 = True
logging.info('Nodemaneger dekommissioniert: {0}'.format(flag3))
break
while True:
time.sleep(5)
status2 = self.ambari.maintenance_on(hostname)
if status2 == 'Request accepted' or status2 == 500:
flag2 = True
logging.info('Wartungsanfrage akzeptiert: {0}'.format(flag2))
break
while True:
time.sleep(5)
status4 = self.ambari.check_maintenance(hostname, 'NODEMANAGER')
if status4 == 'ON' or status4 == 'IMPLIED_FROM_HOST':
flag4 = True
self.ambari.stop_all_services(hostname)
logging.info('Wartung aktiv: {0}'.format(flag4))
logging.info('Dienstleistungen werden gestoppt')
break
time.sleep(90)
status5 = self.mcs.vm_turn_off(hostname)
while True:
time.sleep(5)
status5 = self.mcs.get_vm_info(hostname)['server']['status']
if status5 == 'SHUTOFF':
flag5 = True
logging.info('VM ist ausgeschaltet: {0}'.format(flag5))
break
if flag1 and flag2 and flag3 and flag4 and flag5:
message = 'Erfolg'
logging.info('Herunterskalierung abgeschlossen')
logging.info('Abkühlphase hat begonnen. Warten Sie einige Minuten')
return messageWir nehmen Klassen als Eingabe entgegen Ambari enthält. und Mcs, eine Liste von Knoten, die für das Scaling erlaubt sind, und die Konfigurationsparameter der Knoten: Speicher und CPU, die einem Knoten in YARN zugewiesen sind. Außerdem gibt es 2 interne Parameter q_ram, q_cpu, die Queues sind. Mit deren Hilfe speichern wir die aktuellen Belastungswerte des Clusters. Wenn wir sehen, dass in den letzten 5 Minuten konstant eine hohe Last war, entscheiden wir, dass wir +1 Knoten zum Cluster hinzufügen müssen. Gleiches gilt für den Zustand einer Unterbelastung des Clusters.
Im obigen Code ist ein Beispiel für eine Funktion aufgeführt, die eine Maschine aus dem Cluster entfernt und sie in der Cloud ausschaltet. Zunächst findet die Dekommissionierung statt YARN Nodemanager, dann wird der Modus aktiviert Wartung, dann stoppen wir alle Dienste auf der Maschine und schalten die virtuelle Maschine in der Cloud aus.
2. Skript observer.py
Beispielcode von dort:
if scaler.assert_up(config.scale_up_thresholds) == True:
hostname = cloud.get_vm_to_up(config.scaling_hosts)
if hostname != None:
status1 = scaler.scale_up(hostname)
if status1 == 'Success':
text = {"text": "{0} wurde erfolgreich hochskaliert".format(hostname)}
post = {"text": "{0}".format(text)}
json_data = json.dumps(post)
req = requests.post(webhook, data=json_data.encode('ascii'), headers={'Content-Type': 'application/json'})
time.sleep(config.cooldown_period*60)Hier prüfen wir, ob die Bedingungen für das Aufstocken der Cluster-Ressourcen erfüllt sind und ob Reserve-VMs verfügbar sind, holen den Hostnamen einer von ihnen, fügen sie dem Cluster hinzu und veröffentlichen eine Nachricht darüber in Slack unseres Teams. Danach startet cooldown_period, in dem wir nichts zum Cluster hinzufügen oder entfernen, sondern einfach die Auslastung überwachen. Stabilisiert sich diese und bleibt innerhalb des optimalen Wertebereichs, setzen wir die Überwachung fort. Wenn jedoch eine Node nicht ausreicht, fügen wir eine weitere hinzu.
In Fällen, in denen eine Veranstaltung bevorsteht, wissen wir bereits, dass eine Node nicht ausreicht, daher starten wir sofort alle freien Nodes und halten sie bis zum Ende der Veranstaltung aktiv. Dies geschieht mit Hilfe einer Liste von Zeitstempeln der Veranstaltungen.
Fazit
Der Autoscaler ist eine gute und praktische Lösung für Fälle, in denen eine ungleiche Auslastung des Clusters vorliegt. Gleichzeitig erreichen Sie die erforderliche Cluster-Konfiguration für Spitzenlasten, ohne das Cluster während der Unterauslastung aufrechtzuerhalten, und sparen Kosten. Außerdem geschieht all dies automatisiert ohne Ihr Eingreifen. Der Autoscaler selbst ist nicht mehr als eine Sammlung von Anfragen an die API des Cluster-Managers und die API des Cloud-Anbieters, die nach einer bestimmten Logik verfasst sind. Man sollte sich definitiv an die Unterteilung der Nodes in 3 Typen erinnern, wie wir vorher beschrieben haben. Und das wird Ihnen Glück bringen.
Quelle: habr.com
