Hallo! Wij leren mensen werken met big data. Het is ondenkbaar om een opleidingsprogramma voor big data voor te stellen zonder je eigen cluster, waar alle deelnemers samen aan werken. Om deze reden is het er altijd in ons programma 🙂 We zijn bezig met de configuratie, tuning en administratie, en de studenten voeren daar MapReduce-taken uit en gebruiken Spark.
In dit bericht vertellen we hoe we het probleem van onevenredige belasting van het cluster hebben opgelost door onze eigen autoscaler te schrijven, gebruikmakend van de cloud. .
Probleem
Ons cluster wordt niet helemaal in een typische modus gebruikt. De benutting is erg oneven. Bijvoorbeeld, er zijn praktische lessen waarin alle 30 studenten en de docent toegang krijgen tot het cluster en het beginnen te gebruiken. Of weer zijn er dagen voor de deadline waarin de belasting sterk toeneemt. Het meeste van de tijd werkt het cluster in een onderbenuttigde modus.
Oplossing №1 – houd een cluster dat piekbelastingen aankan, maar gedurende de rest van de tijd ongebruikt blijft.
Oplossing №2 – houd een klein cluster en voeg handmatig knooppunten toe voor de lessen en tijdens piekbelastingen.
Oplossing №3 – houd een klein cluster en schrijf een autoscaler die de huidige belasting van het cluster in de gaten houdt en automatisch, gebruikmakend van verschillende API's, knooppunten aan het cluster toevoegt en verwijdert.
In dit bericht zullen we het over oplossing №3 hebben. Zo'n autoscaler hangt sterk af van externe factoren en niet van interne, en aanbieders bieden dit vaak niet aan. Wij maken gebruik van de cloudinfrastructuur van Mail.ru Cloud Solutions en hebben een autoscaler geschreven met behulp van de MCS API. Aangezien we mensen opleiden in het werken met gegevens, besloten we te laten zien hoe je zo'n autoscaler kunt schrijven voor je eigen doelen en deze met je eigen cloud kunt gebruiken.
Vereisten
Ten eerste moet je een Hadoop-cluster hebben. Wij gebruiken bijvoorbeeld de distributie HDP.
Om ervoor te zorgen dat knooppunten snel kunnen worden toegevoegd en verwijderd, moet er een bepaalde rolverdeling zijn over de knooppunten.
- Master-knooppunt. Hier hoeft niet veel uitleg bij: het belangrijkste knooppunt van het cluster, waarop bijvoorbeeld de Spark-driver draait, als je gebruikmaakt van de interactieve modus.
- Datum-knooppunt. Dit is het knooppunt waarop je gegevens op HDFS worden opgeslagen en waar de berekeningen plaatsvinden.
- Rekenknooppunt. Dit is het knooppunt waarop je niets op HDFS opslaat, maar waar de berekeningen plaatsvinden.
Belangrijk punt. Autoscaling gebeurt ten koste van knooppunten van het derde type. Als je knooppunten van het tweede type begint te verwijderen of toe te voegen, zal de reactietijd aanzienlijk lager zijn - het decommissies en recommissies zal uren in beslag nemen voor jouw cluster. Dit is natuurlijk niet wat je verwacht van autoscaling. Dat wil zeggen, we blijven af van de knooppunten van het eerste en tweede type. Zij zullen een minimaal levensvatbaar cluster vormen dat zal bestaan gedurende de hele looptijd van het programma.
Dus, onze autoscaler is geschreven in Python 3, gebruikt de Ambari API voor het beheer van de cluster services, gebruikt (MCS) voor het starten en stoppen van machines.
Architectuur van de oplossing
- Module
autoscaler.py. Het bevat drie klassen: 1) functies voor het werken met Ambari, 2) functies voor het werken met MCS, 3) functies die rechtstreeks verband houden met de logica van de autoscaler. - Script
observer.py. Het bestaat in wezen uit verschillende regels: wanneer en op welke momenten de functies van de autoscaler moeten worden aangeroepen. - Bestand met configuratieparameters
config.py. Hierin bevindt zich bijvoorbeeld een lijst van knooppunten die zijn toegestaan voor autoscaling en andere parameters die invloed hebben op bijvoorbeeld hoe lang te wachten vanaf het moment dat een nieuw knooppunt is toegevoegd. Hier vind je ook de tijdstempels voor de start van lessen, zodat de maximaal toegestane clusterconfiguratie vóór de les kan worden gestart.
Laten we nu eens kijken naar stukken code in de eerste twee bestanden.
1. Module autoscaler.py
Klas Ambari
Dit is een fragment van code dat de klasse bevat Ambari:
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 = 'Request accepted'
else:
message = req.status_code
return messageBovenaan is er een voorbeeld van de implementatie van de functie stop_all_services, die alle services op de gewenste node in de cluster stopt.
In de invoer van de klasse Ambari geef je de volgende parameters door:
ambari_url, bijvoorbeeld in de volgende vorm'http://localhost:8080/api/v1/clusters/',cluster_name– de naam van je cluster in Ambari,headers = {'X-Requested-By': 'ambari'}- en daarin
authstaan je gebruikersnaam en wachtwoord voor Ambari:auth = ('login', 'password').
De functie zelf omvat niet meer dan een paar aanroepen via de REST API naar Ambari. Logisch gezien krijgen we eerst een lijst van draaiende services op de node en vragen vervolgens om die services in de cluster op die node over te zetten naar de status INSTALLED. Functies voor het starten van alle services, voor het overzetten van nodes naar de status Onderhoud enz. zien er op een soortgelijke manier uit – het zijn gewoon een paar aanvragen via de API.
Klas Mcs
Dit is een fragment van code dat de klasse bevat 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_codeIn de invoer van de klasse Mcs geven we de project-ID binnen de cloud en de gebruikers-ID door, evenals zijn wachtwoord. In de functie vm_turn_on We willen een van de machines inschakelen. De logica hier is iets complexer. Aan het begin van de code wordt een oproep gedaan naar drie andere functies: 1) we moeten een token verkrijgen, 2) we moeten de hostname omzetten naar de namen van de machine in MCS, 3) het id van deze machine verkrijgen. Vervolgens doen we gewoon een post-verzoek en starten we deze machine op.
Zo ziet de functie voor het verkrijgen van het token eruit:
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.tokenDe klasse Autoscaler
In deze klasse bevinden zich functies die betrekking hebben op de werklogica.
Zo ziet een stuk code van deze klasse eruit:
klasse Autoscaler:
def __init__(self, ambari, mcs, schaal_hosts, yarn_ram_per_node, yarn_cpu_per_node):
self.schaal_hosts = schaal_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 schaal_naar_beneden(self, hostname):
vlag1 = vlag2 = vlag3 = vlag4 = vlag5 = False
if hostname in self.schaal_hosts:
while True:
time.sleep(5)
status1 = self.ambari.decommission_nodemanager(hostname)
if status1 == 'Verzoek geaccepteerd' of status1 == 500:
vlag1 = True
logging.info('Dekommissieverzoek geaccepteerd: {0}'.format(vlag1))
break
while True:
time.sleep(5)
status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
if status3 == 'GEÏNSTALLEERD':
vlag3 = True
logging.info('Nodemaneger gedekommissioneerd: {0}'.format(vlag3))
break
while True:
time.sleep(5)
status2 = self.ambari.maintenance_on(hostname)
if status2 == 'Verzoek geaccepteerd' of status2 == 500:
vlag2 = True
logging.info('Onderhoudsverzoek geaccepteerd: {0}'.format(vlag2))
break
while True:
time.sleep(5)
status4 = self.ambari.check_maintenance(hostname, 'NODEMANAGER')
if status4 == 'AAN' of status4 == 'IMPLICIET_VAN_HOST':
vlag4 = True
self.ambari.stop_all_services(hostname)
logging.info('Onderhoud is aan: {0}'.format(vlag4))
logging.info('Stoppen van diensten')
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 == 'UITGESCHAKELD':
vlag5 = True
logging.info('VM is uitgeschakeld: {0}'.format(vlag5))
break
if vlag1 and vlag2 and vlag3 and vlag4 and vlag5:
boodschap = 'Succes'
logging.info('Schaalverkleining voltooid')
logging.info('Afkoelperiode is gestart. Wacht enkele minuten')
return boodschapAls input nemen we klassen Ambari en Mcs, een lijst van knooppunten die zijn toegestaan voor schaalvergroting, evenals de configuratieparameters van knooppunten: geheugen en cpu toegewezen aan een knooppunt in YARN. Daarnaast zijn er 2 interne parameters q_ram, q_cpu, die als wachtrijen functioneren. Hiermee slaan we de waarden van de huidige belasting van het cluster op. Als we zien dat er de afgelopen 5 minuten constant een hoge belasting was, besluiten we dat er +1 knooppunt aan het cluster moet worden toegevoegd. Hetzelfde geldt voor de toestand van een onderbelast cluster.
In de bovenstaande code is een voorbeeld van een functie weergegeven die een machine uit het cluster verwijdert en deze in de cloud stopt. Eerst vindt de decommission plaats YARN Nodemanager, daarna schakelen we de modus in Onderhoud, vervolgens stoppen we alle services op de machine en schakelen we de virtuele machine in de cloud uit.
2. Script observer.py
Voorbeeldcode van daar:
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} is succesvol geschaald".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 controleren we of de voorwaarden zijn vervuld om de capaciteit van het cluster te verhogen en of er überhaupt reserve machines zijn, we verkrijgen de hostnaam van een van hen, voegen deze toe aan het cluster en publiceren een bericht hierover in Slack voor ons team. Daarna wordt cooldown_period, waarin we niets aan het cluster toevoegen of verwijderen, maar gewoon de belasting monitoren. Als deze stabilizeert en binnen het optimale belastingbereik ligt, blijven we gewoon monitoren. Als één node niet genoeg is, voegen we er nog een toe.
Voor gevallen waarin we een activiteit voor de boeg hebben, weten we al zeker dat één node niet genoeg zal zijn, dus starten we meteen alle beschikbare nodes en houden we ze actief tot het einde van de activiteit. Dit gebeurt met behulp van een lijst van tijdstempels van de activiteiten.
Conclusie
Autoscaler is een goede en handige oplossing voor gevallen waarin je een ongelijke belasting van het cluster hebt. Je bereikt tegelijkertijd de vereiste configuratie van het cluster voor piekbelasting en houdt dit cluster niet in stand tijdens onderbelasting, wat kosten bespaart. Bovendien gebeurt dit alles geautomatiseerd zonder jouw tussenkomst. De autoscaler zelf is niet meer dan een set verzoeken aan de API van de clusterbeheerder en de API van de cloudprovider, gestructureerd volgens een bepaalde logica. Waar je zeker aan moet denken, is de indeling van nodes in 3 types, zoals we eerder schreven. En dan zal je gelukkig zijn.
Bron: habr.com
