PĂ«rshĂ«ndetje! Ne e mĂ«sojmĂ« njerĂ«zit si tĂ« punojnĂ« me tĂ« dhĂ«nat e mĂ«dha. Nuk Ă«shtĂ« e mundur tĂ« imagjinohet njĂ« program edukativ pĂ«r tĂ« dhĂ«nat e mĂ«dha pa njĂ« klaster tĂ« vetin, ku tĂ« gjithĂ« pjesĂ«marrĂ«sit punojnĂ« sĂ« bashku. PĂ«r kĂ«tĂ« arsye, nĂ« programin tonĂ« ai Ă«shtĂ« gjithmonĂ« aty đ Ne merremi me konfigurimin, tuningun dhe administrimin e tij, ndĂ«rsa djemtĂ« e pĂ«rdorin atĂ« pĂ«r tĂ« nisur punĂ«t MapReduce dhe pĂ«r tĂ« shfrytĂ«zuar Spark-in.
Në këtë post, ne do të flasim se si e zgjidhëm problemin e ngarkesës së pabarabartë të klasterit duke shkruar një auto-skalues, duke përdorur cloud. .
Problemi
Klasteri ynë nuk përdoret krejtësisht në një mënyrë tipike. Shfrytëzimi është shumë i pabarabartë. Për shembull, ka seanca praktike të cilat, kur të gjithë 30 persona dhe mësuesi hyjnë në klaster dhe fillojnë ta përdorin. Ose ka ditë përpara deadlines, kur ngarkesa rritet shumë. Në çdo kohë tjetër klasteri punon në një mod është nën ngarkesë.
Zgjidhja nr. 1 â Ă«shtĂ« tĂ« mbani njĂ« klaster qĂ« do tĂ« pĂ«rballojĂ« ngarkesat maksimale, por do tĂ« mbetet i papunĂ« nĂ« çdo kohĂ« tjetĂ«r.
Zgjidhja nr. 2 â Ă«shtĂ« tĂ« mbani njĂ« klaster tĂ« vogĂ«l, nĂ« tĂ« cilin do tĂ« shtoni manualisht node pĂ«rpara seancave dhe gjatĂ« ngarkesave maksimale.
Zgjidhja nr. 3 â Ă«shtĂ« tĂ« mbani njĂ« klaster tĂ« vogĂ«l dhe tĂ« shkruani njĂ« auto-skalues, i cili do tĂ« monitorojĂ« ngarkesĂ«n aktuale tĂ« klasterit dhe vetĂ«, duke pĂ«rdorur API tĂ« ndryshme, do tĂ« shtojĂ« dhe heqĂ« node nga klasteri.
Në këtë post ne do të flasim për zgjidhjen nr. 3. Ky auto-skalues varet shumë nga faktorët e jashtëm, e jo nga ata të brendshëm, dhe ofruesit shpesh nuk e ofrojnë atë. Ne përdorim infrastruktura cloud të Mail.ru Cloud Solutions dhe kemi shkruar një auto-skalues, duke përdorur API MCS. Dhe meqenëse ne e mësojmë si të punojmë me të dhënat, vendosëm të tregojmë se si mund të shkruani një auto-skalues të tillë për qëllimet tuaja dhe ta përdorni me cloud-in tuaj.
Kushtet paraprake
Së pari, ju duhet të keni një klaster Hadoop. Ne, për shembull, po përdorim shpërndarjen HDP.
Që node-t tuaja të mund të shtohen dhe hiqen shpejt, ju duhet një shpërndarje e caktuar e rolit për node.
- Master node. Këtu nuk ka nevojë për shumë shpjegim: node kryesore e klasterit, në të cilën funksionon, për shembull, drejtuesi i Spark-it, nëse po përdorni modalitetin interaktiv.
- Data node. Ky është node ku keni të dhënat e ruajtura në HDFS dhe aty ndodhin edhe llogaritë.
- Noda llogarike. Kjo është një node ku nuk mbani asgjë në HDFS, por ku ndodhin llogaritjet.
Moment i rëndësishëm. Autoskalimi do të ndodhë përmes nodave të tretë. Nëse filloni të tërheqni dhe shtoni nodat e tipit të dytë, atëherë shpejtësia e reagimit do të jetë shumë e ulët - dekomishen dhe rekomishen do të marrë orë në klasterin tuaj. Kjo, natyrisht, nuk është ajo që prisni nga autoskalimi. Pra, ne nuk prekim nodat e tipit të parë dhe të dytë. Ato do të përfaqësojnë një klaster minimalisht të jetueshëm, i cili do të ekzistojë gjatë gjithë kohës së programit.
Pra, autoskaleri ynë është shkruar në Python 3, përdor API-në e Ambari për menaxhimin e shërbimeve të klasterit, përdor (MCS) për të nisur dhe ndaluar makinat.
Arkitektura e zgjidhjes
- Moduli
autoscaler.py. Në të janë të shkruara tri klasa: 1) funksione për punë me Ambari, 2) funksione për punë me MCS, 3) funksione të lidhura drejtpërdrejt me logjikën e funksionimit të autoskaluesit. - Script
observer.py. Në thelb përbëhet nga rregulla të ndryshme: kur dhe në cilat momente të thirren funksionet e autoskaluesit. - Skedari me parametrat e konfigurimit
config.py. Aty përmban, për shembull, listën e nodave të lejuara për autoskalim dhe parametrat e tjerë që ndikojnë, për shembull, në sa kohë të presim nga momenti kur është shtuar një node e re. Po ashtu aty ndodhen dhe timestampet e fillimit të mbledhjeve, në mënyrë që para mbledhjes të jetë aktivizuar konfigurimi maksimal i lejuar i klasterit.
Tani le të shikojmë copat e kodit që ndodhen brenda dy skedarëve të parë.
1. Moduli autoscaler.py
Klasa Ambari
Kështu duket një copë kodi që përmban klasën Ambari:
klasa 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 messageMund të shihni implementimin e funksionit më sipër si shembull stop_all_services, i cili ndalon të gjitha shërbimet në nodën e nevojshme të klasterit.
Në klasë Ambari i jepni si_input:
ambari_url, pĂ«r shembull, nĂ« formatin e'http://localhost:8080/api/v1/clusters/',cluster_nameâ emri i klasterit tuaj nĂ« Ambari,headers = {'X-Requested-By': 'ambari'}- dhe brenda
authndodhet emri i përdoruesit dhe fjalëkalimi për Ambari:auth = ('login', 'password').
Funksioni vetë përbën jo më shumë se disa thirrje përmes REST API për Ambari. Nga pikëpamja logjike, fillimisht marrim një listë të shërbimeve të nisura në nodë, dhe pastaj i kërkojmë në këtë klaster, në këtë nodë, që shërbimet në listë të kalojnë në gjendjen INSTALLED. Funksionet për nisjen e të gjitha shërbimeve, për kalimin e nodave në gjendjen Maintenance dhe të tjerët duken gjithashtu të ngjashme - thjesht janë disa kërkesa përmes API-së.
klasa Mcs
Kështu duket një copë kodi që përmban klasën Mcs:
klasa 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_codeNë klasë Mcs ne kalojmë id-në e projektit brenda cloud-it dhe id-në e përdoruesit, si dhe fjalëkalimin e tij. Në funksionin vm_turn_on Ne duam të aktivizojmë një nga makinat. Logjika këtu është pak më komplekse. Në fillim të kodit ka thirrje për tre funksione të tjera: 1) na nevojitet të marrim token-in, 2) na nevojitet të konvertojmë hostname në emrin e makinës në MCS, 3) të marrim id-në e kësaj makine. Më pas, ne bëjmë thjesht një post-request dhe nisi këtë makinë.
Kjo është funksioni për marrjen e token-it:
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.tokenKlasa Autoscaler
Në këtë klasë janë përfshira funksionet që lidhen me logjikën e punës.
Kjo është një pjesë e kodit të kësaj klase:
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('Decomission request accepted: {0}'.format(flag1))
break
while True:
time.sleep(5)
status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
if status3 == 'INSTALLED':
flag3 = True
logging.info('Nodemaneger decommissioned: {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('Maintenance request accepted: {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('Maintenance is on: {0}'.format(flag4))
logging.info('Stopping services')
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 is turned off: {0}'.format(flag5))
break
if flag1 and flag2 and flag3 and flag4 and flag5:
message = 'Success'
logging.info('Scale-down finished')
logging.info('Cooldown period has started. Wait for several minutes')
return messageNe input, ne marrim klaset Ambari dhe Mcs, lista e nodave që janë të lejuara për skalimin, si dhe parametrat e konfigurimit të nodave: RAM dhe CPU të rezervuara për nodin në YARN. Ka gjithashtu 2 parameter të brendshëm q_ram, q_cpu, që janë radhë. Me ndihmën e tyre ruajmë vlerat e ngarkesës aktuale të grumbullit. Nëse shohim që gjatë 5 minutave të fundit ka qenë stabilisht një ngarkesë e lartë, ne vendosim se duhet të shtojmë +1 nod në grumbull. E njëjta gjë vlen edhe për gjendjen e pakufizuar të grumbullit.
Në kodin e mësipërm, jepet një shembull funksioni që heq makinën nga grumbulli dhe e ndalon atë në cloud. Fillimisht ndodh dekoma YARN Nodemanager, më pas aktivizohet moda Maintenance, më pas ndalim të gjitha shërbimet në makinë dhe fikim makinën virtuale në cloud.
2. Skripti observer.py
Shembuj kodi nga atje:
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} është rritur me sukses".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)Në të kjo ne kontrollojmë nëse janë plotësuar kushtet për rritjen e kapaciteteve të klasterit dhe nëse ka ndonjë makinë të lirë, marrim emrin e host-it të njërës prej tyre, e shtojmë atë në klaster dhe publikojmë një mesazh për këtë në Slack të ekipit tonë. Pas kësaj fillon cooldown_period, kur ne nuk shtojmë dhe nuk heqim asgjë nga klasteri, por thjesht monitorojmë ngarkesën. Nëse ajo është stabilizuar dhe është brenda koridorit optimal të vlerave të ngarkesës, thjesht vazhdojmë monitorimin. Nëse një nodë nuk mjafton, atëherë shtojmë një tjetër.
Për rastet kur kemi një aktivitet përpara, ne tashmë e dimë me siguri që një nodë nuk do të mjaftojë, prandaj menjëherë fillojmë të gjitha nodet e lira dhe i mbajmë të aktivizuara deri në fund të aktivitetit. Kjo ndodh përmes një liste të timestamp-ëve të aktiviteteve.
PĂ«rfundim
Autoskaler-i është një zgjidhje e mirë dhe e përshtatshme për ato raste kur keni ngarkesë të pakoordinuar në klaster. Ju arrini një konfigurim të nevojshëm të klasterit për ngarkesa maksimale dhe për më tepër nuk e mbani këtë klaster gjatë periudhave të nën-ngarkesës, duke kursyer fonde. Dhe gjithashtu, kjo ndodh në mënyrë automatike pa pjesëmarrjen tuaj. Vetë autoskaler-i është thjesht një grup kërkesash në API-në e menaxherit të klasterit dhe API-në e ofruesit të cloud-it, të shkruara sipas një logjike të caktuar. Ajo që duhet të mbani mend me siguri është ndarja e nodëve në 3 lloje, siç kemi shkruar më parë. Dhe do të jeni të lumtur.
Burimi: habr.com
